Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,14 @@ jobs:
- name: Compile Python scripts
run: python -m compileall wordpress-api-pro/scripts

- name: Unit tests
run: python3 tests/test_cpt_seeding.py

- name: Seed dry-run smoke (no network)
run: |
python3 wordpress-api-pro/scripts/seed_content.py --dataset tests/fixtures/seed.json > /tmp/plan.json
python3 -c "import json;d=json.load(open('/tmp/plan.json'));assert d['dry_run'] and len(d['plan'])==2;print('smoke ok')"

- name: Run package smoke tests
run: npm test

Expand Down
10 changes: 10 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,15 @@
# Changelog

## 3.6.0 - 2026-06-02

CPT content seeding (Tier-1 dynamic content).

- `create_post.py` gains `--post-type` (resolves rest_base via `/wp/v2/types`) and `--terms` (name→id, create-missing); now importable.
- New `describe_cpt.py` — read-only schema discovery (rest_base, taxonomies, sampled field keys).
- New `seed_content.py` — batch-create CPT entries with ACF/Jet fields, taxonomies, and featured images from a JSON dataset. **Dry-run by default**; `--execute` to write; per-entry errors collected, batch continues. Dry-run/planning is stdlib-only (write-path deps imported lazily).
- `upload_media.py` made importable (`__main__` guard).
- CI runs new unit tests + an offline dry-run smoke.

## 3.5.1 - 2026-06-01

ClawHub packaging compatibility.
Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "wordpress-api-pro",
"version": "3.5.1",
"version": "3.6.0",
"description": "WordPress REST API integration skill for OpenClaw - manage posts, pages, media, WooCommerce, Elementor, and metadata with explicit safety boundaries",
"private": true,
"main": "wordpress-api-pro/SKILL.md",
Expand Down
21 changes: 21 additions & 0 deletions tests/fixtures/seed.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
[
{
"post_type": "projects",
"title": "Acme Rebrand",
"content": "<p>Full brand refresh.</p>",
"status": "draft",
"terms": { "project_category": ["Branding"] },
"featured_image": 42,
"acf": { "client": "Acme", "year": 2025 },
"jet": { "duration_weeks": 6 }
},
{
"post_type": "projects",
"title": "Globex Site",
"content": "<p>Marketing site.</p>",
"status": "draft",
"terms": { "project_category": ["Web"] },
"featured_image": "https://example.com/globex.jpg",
"acf": { "client": "Globex", "year": 2024 }
}
]
65 changes: 65 additions & 0 deletions tests/test_cpt_seeding.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
import json, os, sys, unittest
from unittest import mock

SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "wordpress-api-pro", "scripts")
sys.path.insert(0, os.path.abspath(SCRIPTS))

import create_post # noqa: E402


class FakeResp:
def __init__(self, payload, code=200):
self._b = json.dumps(payload).encode()
self.status = code
def read(self): return self._b
def __enter__(self): return self
def __exit__(self, *a): return False


class ResolveRestBaseTest(unittest.TestCase):
def test_uses_rest_base_from_types(self):
with mock.patch.object(create_post.urllib.request, "urlopen",
return_value=FakeResp({"rest_base": "projects"})):
self.assertEqual(
create_post.resolve_rest_base("http://x", "a", "projects"), "projects")

def test_falls_back_to_slug_on_error(self):
with mock.patch.object(create_post.urllib.request, "urlopen",
side_effect=Exception("404")):
self.assertEqual(
create_post.resolve_rest_base("http://x", "a", "team"), "team")


class ResolveTermsTest(unittest.TestCase):
def test_existing_term_resolves_to_id(self):
responses = [
FakeResp({"rest_base": "project_category"}), # taxonomy rest base
FakeResp([{"id": 5, "name": "Branding"}]), # term search hit
]
with mock.patch.object(create_post.urllib.request, "urlopen",
side_effect=responses):
out = create_post.resolve_terms("http://x", "a",
{"project_category": ["Branding"]},
create_missing=False)
self.assertEqual(out, {"project_category": [5]})


import seed_content # noqa: E402


class SeedDryRunTest(unittest.TestCase):
def test_dry_run_plans_every_entry_without_network(self):
fixture = os.path.join(os.path.dirname(__file__), "fixtures", "seed.json")
with open(fixture) as f:
dataset = json.load(f)
plan = seed_content.plan_seed(dataset)
self.assertEqual(len(plan), 2)
self.assertEqual(plan[0]["post_type"], "projects")
self.assertIn("acf", plan[0]["will_set"])
self.assertIn("terms", plan[0]["will_set"])
self.assertEqual(plan[1]["featured_image_kind"], "url")
self.assertEqual(plan[0]["featured_image_kind"], "media_id")


if __name__ == "__main__":
unittest.main()
18 changes: 17 additions & 1 deletion wordpress-api-pro/SKILL.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
---
name: wordpress-api-pro
version: 3.5.1
version: 3.6.0
license: MIT-0
description: |
WordPress REST API integration for managing posts, pages, media, WooCommerce products, Elementor content, SEO meta, ACF, and JetEngine fields.
Expand Down Expand Up @@ -199,9 +199,25 @@ python3 scripts/upload_media.py \
- `scripts/acf_fields.py` — read/write ACF fields.
- `scripts/seo_meta.py` — read/write Rank Math and Yoast SEO metadata.
- `scripts/jetengine_fields.py` — read/write JetEngine custom fields.
- `scripts/describe_cpt.py` — discover a CPT's rest_base, taxonomies, and field keys (read-only).
- `scripts/seed_content.py` — batch-create CPT entries with ACF/Jet fields, taxonomies, and featured images from a JSON dataset. **Dry-run by default; pass `--execute` to write.**
- `scripts/elementor_content.py` — read/update Elementor `_elementor_data`.
- `scripts/woo_products.py` — manage WooCommerce products.

## Seeding dynamic content (CPT)

For dynamic sites (JetEngine/ACF listings), populate the entries the listings render:

1. `describe_cpt.py --post-type projects` — learn the rest_base, taxonomies, field keys.
2. Write a JSON dataset (array of `{post_type, title, content, status, terms, featured_image, acf, jet}`).
3. `seed_content.py --dataset data.json` — review the dry-run plan (no writes, stdlib-only).
4. `seed_content.py --dataset data.json --execute` — create (drafts by default).

Notes: the CPT, taxonomies, and ACF field-groups must already exist (admin-side).
`featured_image` accepts a media id or a URL/path (URL fetch needs `--allow-remote-url`).
`--execute` needs the `requests` dependency (used by the ACF/Jet writers). Re-running
creates duplicates (no upsert yet).

## Verification before live writes

Before any live mutation:
Expand Down
132 changes: 99 additions & 33 deletions wordpress-api-pro/scripts/create_post.py
Original file line number Diff line number Diff line change
@@ -1,42 +1,108 @@
#!/usr/bin/env python3
"""Create WordPress post via REST API"""
import argparse, json, os, sys, urllib.request
"""Create a WordPress post or CPT entry via REST API (with taxonomy support)."""
import argparse, json, os, sys, urllib.request, urllib.parse
from base64 import b64encode

def create_post(url, username, password, title, content, status='draft', **kwargs):
api_url = f"{url.rstrip('/')}/wp-json/wp/v2/posts"
credentials = f"{username}:{password}".encode('utf-8')
auth_header = b64encode(credentials).decode('ascii')


def _auth(username, password):
return 'Basic ' + b64encode(f"{username}:{password}".encode()).decode()


def _get(url, auth):
req = urllib.request.Request(url, method='GET')
req.add_header('Authorization', auth)
with urllib.request.urlopen(req) as r:
return json.loads(r.read().decode())


def _post(url, auth, payload):
req = urllib.request.Request(url, data=json.dumps(payload).encode(), method='POST')
req.add_header('Authorization', auth)
req.add_header('Content-Type', 'application/json')
with urllib.request.urlopen(req) as r:
return json.loads(r.read().decode())


def resolve_rest_base(base_url, auth, post_type):
"""Resolve a post type's REST base; fall back to the slug on any error."""
try:
info = _get(f"{base_url.rstrip('/')}/wp-json/wp/v2/types/{post_type}", auth)
return info.get('rest_base') or post_type
except Exception:
return post_type


def resolve_terms(base_url, auth, terms_dict, create_missing=True):
"""Map {taxonomy: [name|id, ...]} -> {taxonomy: [id, ...]}.

Names are resolved (and optionally created) via the taxonomy's REST base.
Integer-like values pass through as ids.
"""
base_url = base_url.rstrip('/')
out = {}
for taxonomy, values in (terms_dict or {}).items():
tax_base = resolve_rest_base(base_url, auth, taxonomy) # taxonomy rest_base
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Resolve taxonomy REST bases from taxonomies

When a taxonomy has a custom rest_base, this uses the post-type resolver, which requests /wp/v2/types/{taxonomy} rather than the taxonomy descriptor endpoint. That request fails and falls back to the taxonomy slug, so term search/creation later targets /wp/v2/{taxonomy} and breaks for taxonomies whose REST collection is renamed (the same scenario this rest_base lookup is meant to handle).

Useful? React with 👍 / 👎.

ids = []
for v in values:
if isinstance(v, int) or (isinstance(v, str) and v.isdigit()):
ids.append(int(v)); continue
q = urllib.parse.quote(str(v))
hits = _get(f"{base_url}/wp-json/wp/v2/{tax_base}?search={q}", auth)
match = next((t for t in hits if str(t.get('name', '')).lower() == str(v).lower()), None)
if match:
ids.append(match['id'])
elif create_missing:
created = _post(f"{base_url}/wp-json/wp/v2/{tax_base}", auth, {'name': v})
ids.append(created['id'])
else:
raise ValueError(f"Term '{v}' not found in '{taxonomy}'")
out[taxonomy] = ids
return out


def create_post(url, username, password, title, content, status='draft',
post_type='post', featured_media=None, terms=None):
"""Create a post/CPT entry. Returns the created object dict. Raises on error."""
auth = _auth(username, password)
base = url.rstrip('/')
rest_base = resolve_rest_base(base, auth, post_type)

data = {'title': title, 'content': content, 'status': status}
if 'featured_media' in kwargs and kwargs['featured_media']:
data['featured_media'] = int(kwargs['featured_media'])

request = urllib.request.Request(api_url, data=json.dumps(data).encode('utf-8'), method='POST')
request.add_header('Authorization', f'Basic {auth_header}')
request.add_header('Content-Type', 'application/json')

if featured_media:
data['featured_media'] = int(featured_media)
if terms:
resolved = resolve_terms(base, auth, terms)
for taxonomy, ids in resolved.items():
data[taxonomy] = ids # REST accepts the taxonomy key with term ids

return _post(f"{base}/wp-json/wp/v2/{rest_base}", auth, data)


def main():
p = argparse.ArgumentParser(description='Create WordPress post or CPT entry')
p.add_argument('--url', default=os.getenv('WP_URL') or os.getenv('WP_SITE_URL'))
p.add_argument('--username', default=os.getenv('WP_USERNAME') or os.getenv('WP_USER'))
p.add_argument('--app-password', default=os.getenv('WP_APP_PASSWORD'))
p.add_argument('--title', required=True)
p.add_argument('--content', required=True)
p.add_argument('--status', default='draft', choices=['publish', 'draft', 'pending'])
p.add_argument('--post-type', default='post')
p.add_argument('--featured-media', type=int)
p.add_argument('--terms', help='JSON {"taxonomy": ["Name or id", ...]}')
a = p.parse_args()
if not all([a.url, a.username, a.app_password]):
print(json.dumps({"error": "Missing required credentials"}), file=sys.stderr)
sys.exit(1)
try:
with urllib.request.urlopen(request) as response:
result = json.loads(response.read().decode('utf-8'))
print(json.dumps(result, indent=2))
return result
result = create_post(a.url, a.username, a.app_password, a.title, a.content,
a.status, post_type=a.post_type,
featured_media=a.featured_media,
terms=json.loads(a.terms) if a.terms else None)
print(json.dumps(result, indent=2))
except Exception as e:
print(json.dumps({"error": str(e)}), file=sys.stderr)
sys.exit(1)

parser = argparse.ArgumentParser(description='Create WordPress post')
parser.add_argument('--url', default=os.getenv('WP_URL'))
parser.add_argument('--username', default=os.getenv('WP_USERNAME'))
parser.add_argument('--app-password', default=os.getenv('WP_APP_PASSWORD'))
parser.add_argument('--title', required=True)
parser.add_argument('--content', required=True)
parser.add_argument('--status', default='draft', choices=['publish', 'draft', 'pending'])
parser.add_argument('--featured-media', type=int)

args = parser.parse_args()
if not all([args.url, args.username, args.app_password]):
print(json.dumps({"error": "Missing required credentials"}), file=sys.stderr)
sys.exit(1)

create_post(args.url, args.username, args.app_password, args.title, args.content, args.status, featured_media=args.featured_media)

if __name__ == '__main__':
main()
68 changes: 68 additions & 0 deletions wordpress-api-pro/scripts/describe_cpt.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
#!/usr/bin/env python3
"""Describe a custom post type: rest_base, taxonomies, and discovered field keys.

Read-only. Samples the newest existing entry to surface ACF/meta keys so a caller
knows what to populate when seeding.

Usage:
python3 describe_cpt.py --post-type projects
Env: WP_URL/WP_SITE_URL, WP_USERNAME/WP_USER, WP_APP_PASSWORD
"""
import argparse, json, os, sys, urllib.request
from base64 import b64encode


def _auth(u, p): return 'Basic ' + b64encode(f"{u}:{p}".encode()).decode()


def _get(url, auth):
req = urllib.request.Request(url, method='GET')
req.add_header('Authorization', auth)
with urllib.request.urlopen(req) as r:
return json.loads(r.read().decode())


def describe_cpt(base_url, username, password, post_type):
auth = _auth(username, password)
base = base_url.rstrip('/')
info = _get(f"{base}/wp-json/wp/v2/types/{post_type}", auth)
rest_base = info.get('rest_base') or post_type
taxonomies = info.get('taxonomies', [])

field_keys, sampled_id = [], None
try:
entries = _get(f"{base}/wp-json/wp/v2/{rest_base}?per_page=1&orderby=date", auth)
if entries:
sampled_id = entries[0].get('id')
meta = entries[0].get('meta', {}) or {}
acf = entries[0].get('acf', {}) or {}
keys = set(k for k in meta if not k.startswith('_')) | set(acf.keys())
field_keys = sorted(keys)
except Exception:
pass

return {
'post_type': post_type, 'rest_base': rest_base,
'taxonomies': taxonomies, 'field_keys': field_keys,
'sampled_entry_id': sampled_id,
'note': '' if field_keys else 'No entries to sample; supply field keys manually.',
}


def main():
p = argparse.ArgumentParser(description='Describe a CPT for seeding')
p.add_argument('--url', default=os.getenv('WP_URL') or os.getenv('WP_SITE_URL'))
p.add_argument('--username', default=os.getenv('WP_USERNAME') or os.getenv('WP_USER'))
p.add_argument('--app-password', default=os.getenv('WP_APP_PASSWORD'))
p.add_argument('--post-type', required=True)
a = p.parse_args()
if not all([a.url, a.username, a.app_password]):
print(json.dumps({"error": "Missing required credentials"}), file=sys.stderr); sys.exit(1)
try:
print(json.dumps(describe_cpt(a.url, a.username, a.app_password, a.post_type), indent=2))
except Exception as e:
print(json.dumps({"error": str(e)}), file=sys.stderr); sys.exit(1)


if __name__ == '__main__':
main()
Loading
Loading