Full brand refresh.
", + "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": "Marketing site.
", + "status": "draft", + "terms": { "project_category": ["Web"] }, + "featured_image": "https://example.com/globex.jpg", + "acf": { "client": "Globex", "year": 2024 } + } +] diff --git a/tests/test_cpt_seeding.py b/tests/test_cpt_seeding.py new file mode 100644 index 0000000..8455d92 --- /dev/null +++ b/tests/test_cpt_seeding.py @@ -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() diff --git a/wordpress-api-pro/SKILL.md b/wordpress-api-pro/SKILL.md index c9759b7..68df1db 100644 --- a/wordpress-api-pro/SKILL.md +++ b/wordpress-api-pro/SKILL.md @@ -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. @@ -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: diff --git a/wordpress-api-pro/scripts/create_post.py b/wordpress-api-pro/scripts/create_post.py index 8e6e7a0..1eae334 100755 --- a/wordpress-api-pro/scripts/create_post.py +++ b/wordpress-api-pro/scripts/create_post.py @@ -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 + 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() diff --git a/wordpress-api-pro/scripts/describe_cpt.py b/wordpress-api-pro/scripts/describe_cpt.py new file mode 100644 index 0000000..e45cd86 --- /dev/null +++ b/wordpress-api-pro/scripts/describe_cpt.py @@ -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() diff --git a/wordpress-api-pro/scripts/seed_content.py b/wordpress-api-pro/scripts/seed_content.py new file mode 100644 index 0000000..de0210e --- /dev/null +++ b/wordpress-api-pro/scripts/seed_content.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +"""Seed a dataset of CPT entries with ACF/Jet fields, taxonomies, and images. + +Dry-run by default — validates and prints a per-entry plan with NO writes. +Pass --execute to perform writes. Reuses create_post / acf_fields / +jetengine_fields / upload_media. + +Usage: + python3 seed_content.py --dataset data.json # dry-run (default) + python3 seed_content.py --dataset data.json --execute # write +Env: WP_URL/WP_SITE_URL, WP_USERNAME/WP_USER, WP_APP_PASSWORD +""" +import argparse, json, os, sys + +# NB: the write-path modules (acf_fields/jetengine_fields) import `requests`, and +# the image path needs upload_media. They are imported lazily inside seed() so the +# dry-run / planning path (and tests / CI smoke) need only the stdlib. + + +def plan_seed(dataset): + """Pure planning — no network. Returns a list of per-entry plan dicts.""" + plan = [] + for i, e in enumerate(dataset): + fi = e.get('featured_image') + kind = None + if isinstance(fi, int) or (isinstance(fi, str) and str(fi).isdigit()): + kind = 'media_id' + elif isinstance(fi, str) and fi: + kind = 'url' if fi.lower().startswith(('http://', 'https://')) else 'path' + will_set = [] + if e.get('acf'): will_set.append('acf') + if e.get('jet'): will_set.append('jet') + if e.get('terms'): will_set.append('terms') + if fi is not None: will_set.append('featured_image') + plan.append({ + 'index': i, 'title': e.get('title', '(no title)'), + 'post_type': e.get('post_type', 'post'), + 'status': e.get('status', 'draft'), + 'will_set': will_set, 'featured_image_kind': kind, + }) + return plan + + +def _resolve_image(url, user, pw, fi, allow_remote): + if isinstance(fi, int) or (isinstance(fi, str) and str(fi).isdigit()): + return int(fi) + import upload_media as _media + res = _media.upload_media(url, user, pw, fi, allow_remote_url=allow_remote) + return res.get('id') if isinstance(res, dict) else None + + +def seed(url, user, pw, dataset, allow_remote=False): + """Execute the seed. Returns {created: [...], failed: [...]}.""" + import create_post as _cp + import acf_fields as _acf + import jetengine_fields as _jet + import upload_media as _media + created, failed = [], [] + for e in dataset: + try: + post = _cp.create_post( + url, user, pw, e['title'], e.get('content', ''), + e.get('status', 'draft'), post_type=e.get('post_type', 'post'), + terms=e.get('terms')) + pid = post['id'] + if e.get('acf'): + _acf.set_acf_fields(url, user, pw, pid, e['acf']) + if e.get('jet'): + _jet.set_jetengine_fields(url, user, pw, pid, e['jet']) + if e.get('featured_image') is not None: + mid = _resolve_image(url, user, pw, e['featured_image'], allow_remote) + if mid: + _media.set_featured_image(url, user, pw, pid, mid) + created.append({'id': pid, 'title': e['title']}) + except Exception as ex: + failed.append({'title': e.get('title', '(no title)'), 'error': str(ex)}) + return {'created': created, 'failed': failed} + + +def main(): + p = argparse.ArgumentParser(description='Seed CPT entries from a JSON dataset') + 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('--dataset', required=True, help='Path to JSON array of entries') + p.add_argument('--execute', action='store_true', help='Perform writes (default: dry-run)') + p.add_argument('--allow-remote-url', action='store_true', help='Permit remote image fetches') + a = p.parse_args() + + with open(a.dataset) as f: + dataset = json.load(f) + if not isinstance(dataset, list): + print(json.dumps({"error": "dataset must be a JSON array"}), file=sys.stderr); sys.exit(1) + + if not a.execute: + print(json.dumps({"dry_run": True, "plan": plan_seed(dataset)}, indent=2)) + return + + if not all([a.url, a.username, a.app_password]): + print(json.dumps({"error": "Missing required credentials"}), file=sys.stderr); sys.exit(1) + result = seed(a.url, a.username, a.app_password, dataset, allow_remote=a.allow_remote_url) + print(json.dumps(result, indent=2)) + if result['failed']: + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/wordpress-api-pro/scripts/upload_media.py b/wordpress-api-pro/scripts/upload_media.py index 688f5da..bc59a03 100755 --- a/wordpress-api-pro/scripts/upload_media.py +++ b/wordpress-api-pro/scripts/upload_media.py @@ -118,46 +118,52 @@ def set_featured_image(url, username, app_credential, post_id, media_id): print(json.dumps({"error": str(e)}), file=sys.stderr) sys.exit(1) -parser = argparse.ArgumentParser(description='Upload media to WordPress') -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('--file', required=True, help='Local file path or URL') -parser.add_argument('--title', help='Media title') -parser.add_argument('--alt-text', help='Alt text for images') -parser.add_argument('--caption', help='Media caption') -parser.add_argument('--set-featured', action='store_true', help='Set as featured image') -parser.add_argument('--post-id', type=int, help='Post ID (required with --set-featured)') -parser.add_argument('--allow-remote-url', action='store_true', help='Explicitly allow fetching HTTPS remote media URLs') -args = parser.parse_args() +def main(): + parser = argparse.ArgumentParser(description='Upload media to WordPress') + 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('--file', required=True, help='Local file path or URL') + parser.add_argument('--title', help='Media title') + parser.add_argument('--alt-text', help='Alt text for images') + parser.add_argument('--caption', help='Media caption') + parser.add_argument('--set-featured', action='store_true', help='Set as featured image') + parser.add_argument('--post-id', type=int, help='Post ID (required with --set-featured)') + parser.add_argument('--allow-remote-url', action='store_true', help='Explicitly allow fetching HTTPS remote media URLs') -if not all([args.url, args.username, args.app_password]): - print(json.dumps({"error": "Missing credentials"}), file=sys.stderr) - sys.exit(1) + args = parser.parse_args() -if args.set_featured and not args.post_id: - print(json.dumps({"error": "--post-id required when using --set-featured"}), file=sys.stderr) - sys.exit(1) + if not all([args.url, args.username, args.app_password]): + print(json.dumps({"error": "Missing credentials"}), file=sys.stderr) + sys.exit(1) + + if args.set_featured and not args.post_id: + print(json.dumps({"error": "--post-id required when using --set-featured"}), file=sys.stderr) + sys.exit(1) + + # Upload media + try: + result = upload_media( + args.url, + args.username, + args.app_password, + args.file, + title=args.title, + alt_text=args.alt_text, + caption=args.caption, + allow_remote_url=args.allow_remote_url, + ) + except SafetyError as e: + die_safety(e) + + # Set as featured image if requested + if args.set_featured and 'id' in result: + set_featured_image(args.url, args.username, args.app_password, args.post_id, result['id']) + result['featured_image_set'] = True -# Upload media -try: - result = upload_media( - args.url, - args.username, - args.app_password, - args.file, - title=args.title, - alt_text=args.alt_text, - caption=args.caption, - allow_remote_url=args.allow_remote_url, - ) -except SafetyError as e: - die_safety(e) + print(json.dumps(result, indent=2)) -# Set as featured image if requested -if args.set_featured and 'id' in result: - set_featured_image(args.url, args.username, args.app_password, args.post_id, result['id']) - result['featured_image_set'] = True -print(json.dumps(result, indent=2)) +if __name__ == '__main__': + main()