airano-ir
43fd2201a0
Settings fixes: - MAX_SITES_PER_USER, USER_RATE_LIMIT_PER_MIN/HR now read from DB settings table (DB > ENV > default), so dashboard/settings changes apply without restart. Sync cache refreshed on every save or delete. - /api/me reports the live DB value for max_sites_per_user. Admin improvements: - Admin users bypass per-user rate limiting entirely (role=admin or ADMIN_EMAILS). - Admin Overview now shows platform stats: registered users, new users (7d), total user sites, available tools. Plugin cleanup: - Appwrite and Directus plugins removed from the active registry (8 plugins now: WordPress, WooCommerce, WordPress Specialist, Gitea, n8n, Supabase, OpenPanel, Coolify). Plugin code is retained for future re-enabling. - Settings page plugin visibility list updated to match. Mobile onboarding: - Stepper steps on narrow viewports stack vertically with correct full border and rounded corners on each step. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1206 lines
45 KiB
Python
1206 lines
45 KiB
Python
"""Posts Handler - manages WordPress posts, pages, and custom post types"""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
from plugins.wordpress.client import WordPressClient
|
|
|
|
|
|
def _count_words(html_content: str) -> int:
|
|
"""Strip HTML tags and count words."""
|
|
text = re.sub(r"<[^>]+>", " ", html_content)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
return len(text.split()) if text else 0
|
|
|
|
|
|
def _strip_html(html_content: str, max_chars: int = 500) -> str:
|
|
"""Strip HTML tags and return first max_chars characters."""
|
|
text = re.sub(r"<[^>]+>", " ", html_content)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
if len(text) > max_chars:
|
|
return text[:max_chars] + "..."
|
|
return text
|
|
|
|
|
|
def get_tool_specifications() -> list[dict[str, Any]]:
|
|
"""Return tool specifications for ToolGenerator"""
|
|
return [
|
|
# === POSTS ===
|
|
{
|
|
"name": "list_posts",
|
|
"method_name": "list_posts",
|
|
"description": "List WordPress posts. Returns paginated list of posts with title, excerpt, status, and metadata.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"per_page": {
|
|
"type": "integer",
|
|
"description": "Number of posts per page (1-100)",
|
|
"default": 10,
|
|
"minimum": 1,
|
|
"maximum": 100,
|
|
},
|
|
"page": {
|
|
"type": "integer",
|
|
"description": "Page number",
|
|
"default": 1,
|
|
"minimum": 1,
|
|
},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Filter by post status",
|
|
"enum": ["publish", "draft", "pending", "private", "any"],
|
|
"default": "any",
|
|
},
|
|
"search": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Search term to filter posts",
|
|
},
|
|
"search_terms": {
|
|
"anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}],
|
|
"description": "Multiple search terms to search in parallel. Results are deduplicated. Overrides 'search' if both provided.",
|
|
},
|
|
"include_content": {
|
|
"type": "boolean",
|
|
"description": "Include content summary (first 500 chars) and word count in results. Default false to save tokens.",
|
|
"default": False,
|
|
},
|
|
},
|
|
},
|
|
"scope": "read",
|
|
},
|
|
{
|
|
"name": "get_post",
|
|
"method_name": "get_post",
|
|
"description": "Get a specific WordPress post by ID. Returns complete post data including content, metadata, and author information.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_id": {
|
|
"type": "integer",
|
|
"description": "Post ID to retrieve",
|
|
"minimum": 1,
|
|
},
|
|
"fields": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Comma-separated list of fields to return (e.g., 'id,title,status,tags'). Returns all fields if not specified. Use to reduce response size and token usage.",
|
|
},
|
|
},
|
|
"required": ["post_id"],
|
|
},
|
|
"scope": "read",
|
|
},
|
|
{
|
|
"name": "create_post",
|
|
"method_name": "create_post",
|
|
"description": "Create a new WordPress post. Supports custom slug, categories, tags, and featured image.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {
|
|
"type": "string",
|
|
"description": "Post title (displayed as main heading)",
|
|
"minLength": 1,
|
|
},
|
|
"content": {
|
|
"type": "string",
|
|
"description": "Post content (supports HTML and WordPress blocks)",
|
|
},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Publication status",
|
|
"enum": ["publish", "draft", "pending", "private", "future"],
|
|
"default": "draft",
|
|
},
|
|
"slug": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post URL slug (auto-generated from title if not provided)",
|
|
},
|
|
"excerpt": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post excerpt/summary",
|
|
},
|
|
"categories": {
|
|
"anyOf": [
|
|
{"type": "array", "items": {"type": "integer"}},
|
|
{"type": "null"},
|
|
],
|
|
"description": "Category IDs to assign to post",
|
|
},
|
|
"tags": {
|
|
"anyOf": [
|
|
{"type": "array", "items": {"type": "integer"}},
|
|
{"type": "null"},
|
|
],
|
|
"description": "Tag IDs to assign to post",
|
|
},
|
|
"featured_media": {
|
|
"anyOf": [{"type": "integer"}, {"type": "null"}],
|
|
"description": "Featured image media ID",
|
|
},
|
|
},
|
|
"required": ["title", "content"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
{
|
|
"name": "update_post",
|
|
"method_name": "update_post",
|
|
"description": "Update an existing WordPress post. Can update any field including title, content, status, categories, and tags.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_id": {
|
|
"type": "integer",
|
|
"description": "Post ID to update",
|
|
"minimum": 1,
|
|
},
|
|
"title": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post title",
|
|
},
|
|
"content": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post content",
|
|
},
|
|
"status": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Publication status",
|
|
"enum": ["publish", "draft", "pending", "private", "future"],
|
|
},
|
|
"slug": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post URL slug",
|
|
},
|
|
"excerpt": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Post excerpt",
|
|
},
|
|
"categories": {
|
|
"anyOf": [
|
|
{"type": "array", "items": {"type": "integer"}},
|
|
{"type": "null"},
|
|
],
|
|
"description": "Category IDs",
|
|
},
|
|
"tags": {
|
|
"anyOf": [
|
|
{"type": "array", "items": {"type": "integer"}},
|
|
{"type": "null"},
|
|
],
|
|
"description": "Tag IDs",
|
|
},
|
|
"featured_media": {
|
|
"anyOf": [{"type": "integer"}, {"type": "null"}],
|
|
"description": "Featured image media ID",
|
|
},
|
|
},
|
|
"required": ["post_id"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
{
|
|
"name": "delete_post",
|
|
"method_name": "delete_post",
|
|
"description": "Delete or trash a WordPress post. Can permanently delete or move to trash.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_id": {
|
|
"type": "integer",
|
|
"description": "Post ID to delete",
|
|
"minimum": 1,
|
|
},
|
|
"force": {
|
|
"type": "boolean",
|
|
"description": "Permanently delete (true) or move to trash (false)",
|
|
"default": False,
|
|
},
|
|
},
|
|
"required": ["post_id"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
# === PAGES ===
|
|
{
|
|
"name": "list_pages",
|
|
"method_name": "list_pages",
|
|
"description": "List WordPress pages. Returns paginated list of pages with metadata.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"per_page": {
|
|
"type": "integer",
|
|
"description": "Number of pages per page (1-100)",
|
|
"default": 10,
|
|
"minimum": 1,
|
|
"maximum": 100,
|
|
},
|
|
"page": {
|
|
"type": "integer",
|
|
"description": "Page number",
|
|
"default": 1,
|
|
"minimum": 1,
|
|
},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Filter by page status",
|
|
"enum": ["publish", "draft", "pending", "private", "any"],
|
|
"default": "any",
|
|
},
|
|
},
|
|
},
|
|
"scope": "read",
|
|
},
|
|
{
|
|
"name": "create_page",
|
|
"method_name": "create_page",
|
|
"description": "Create a new WordPress page. Supports HTML content, parent pages, and custom slugs.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {
|
|
"type": "string",
|
|
"description": "Page title (displayed as main heading)",
|
|
"minLength": 1,
|
|
},
|
|
"content": {
|
|
"type": "string",
|
|
"description": "Page content (supports HTML and WordPress blocks)",
|
|
},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Publication status of the page",
|
|
"enum": ["publish", "draft", "pending", "private"],
|
|
"default": "draft",
|
|
},
|
|
"slug": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Page URL slug (e.g., 'about-us'). Auto-generated from title if not provided.",
|
|
},
|
|
"parent": {
|
|
"anyOf": [{"type": "integer"}, {"type": "null"}],
|
|
"description": "Parent page ID for creating hierarchical page structure",
|
|
},
|
|
},
|
|
"required": ["title", "content"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
{
|
|
"name": "update_page",
|
|
"method_name": "update_page",
|
|
"description": "Update an existing WordPress page. Can update title, content, status, slug, parent page, featured image, and post meta (e.g. Yoast fields).",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"page_id": {
|
|
"type": "integer",
|
|
"description": "Page ID to update",
|
|
"minimum": 1,
|
|
},
|
|
"title": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Page title",
|
|
},
|
|
"content": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Page content",
|
|
},
|
|
"status": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Publication status",
|
|
"enum": ["publish", "draft", "pending", "private"],
|
|
},
|
|
"slug": {
|
|
"anyOf": [{"type": "string"}, {"type": "null"}],
|
|
"description": "Page URL slug",
|
|
},
|
|
"parent": {
|
|
"anyOf": [{"type": "integer"}, {"type": "null"}],
|
|
"description": "Parent page ID",
|
|
},
|
|
"featured_media": {
|
|
"anyOf": [{"type": "integer"}, {"type": "null"}],
|
|
"description": "Featured image attachment ID. Pass 0 to clear.",
|
|
},
|
|
"meta": {
|
|
"anyOf": [{"type": "object"}, {"type": "null"}],
|
|
"description": "Post meta key/value map forwarded as REST `meta` (e.g. Yoast fields like `_yoast_wpseo_metadesc`). Only registered/whitelisted keys are persisted by WordPress.",
|
|
},
|
|
},
|
|
"required": ["page_id"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
{
|
|
"name": "delete_page",
|
|
"method_name": "delete_page",
|
|
"description": "Delete or trash a WordPress page. Can permanently delete or move to trash.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"page_id": {
|
|
"type": "integer",
|
|
"description": "Page ID to delete",
|
|
"minimum": 1,
|
|
},
|
|
"force": {
|
|
"type": "boolean",
|
|
"description": "Permanently delete (true) or move to trash (false)",
|
|
"default": False,
|
|
},
|
|
},
|
|
"required": ["page_id"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
# === POST TYPES ===
|
|
{
|
|
"name": "list_post_types",
|
|
"method_name": "list_post_types",
|
|
"description": "List all registered post types including built-in (post, page) and custom post types (portfolio, testimonials, etc.).",
|
|
"schema": {"type": "object", "properties": {}},
|
|
"scope": "read",
|
|
},
|
|
{
|
|
"name": "get_post_type_info",
|
|
"method_name": "get_post_type_info",
|
|
"description": "Get detailed information about a specific post type including supported features and REST API configuration.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_type": {
|
|
"type": "string",
|
|
"description": "Post type slug (e.g., 'portfolio', 'post', 'page')",
|
|
}
|
|
},
|
|
"required": ["post_type"],
|
|
},
|
|
"scope": "read",
|
|
},
|
|
# === CUSTOM POSTS ===
|
|
{
|
|
"name": "list_custom_posts",
|
|
"method_name": "list_custom_posts",
|
|
"description": "List posts of a specific custom post type. Use list_post_types first to discover available post types.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_type": {
|
|
"type": "string",
|
|
"description": "Post type slug (e.g., 'portfolio', 'testimonials')",
|
|
},
|
|
"per_page": {
|
|
"type": "integer",
|
|
"description": "Number of posts per page",
|
|
"default": 10,
|
|
"minimum": 1,
|
|
"maximum": 100,
|
|
},
|
|
"page": {
|
|
"type": "integer",
|
|
"description": "Page number",
|
|
"default": 1,
|
|
"minimum": 1,
|
|
},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Post status filter",
|
|
"enum": ["publish", "draft", "pending", "private", "any"],
|
|
"default": "any",
|
|
},
|
|
},
|
|
"required": ["post_type"],
|
|
},
|
|
"scope": "read",
|
|
},
|
|
{
|
|
"name": "create_custom_post",
|
|
"method_name": "create_custom_post",
|
|
"description": "Create a new post of a custom post type. Supports custom fields/meta data.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_type": {
|
|
"type": "string",
|
|
"description": "Post type slug (e.g., 'portfolio')",
|
|
},
|
|
"title": {"type": "string", "description": "Post title", "minLength": 1},
|
|
"content": {"type": "string", "description": "Post content (HTML allowed)"},
|
|
"status": {
|
|
"type": "string",
|
|
"description": "Post status",
|
|
"enum": ["publish", "draft", "pending", "private", "future"],
|
|
"default": "draft",
|
|
},
|
|
"meta": {
|
|
"anyOf": [{"type": "object"}, {"type": "null"}],
|
|
"description": "Custom fields/meta data",
|
|
},
|
|
},
|
|
"required": ["post_type", "title", "content"],
|
|
},
|
|
"scope": "write",
|
|
},
|
|
# === INTERNAL LINKS ===
|
|
{
|
|
"name": "get_internal_links",
|
|
"method_name": "get_internal_links",
|
|
"description": "Extract internal links from a post's content. Returns list of internal links with anchor text and URL. Useful for SEO internal linking analysis and avoiding duplicate links.",
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"post_id": {
|
|
"type": "integer",
|
|
"description": "Post ID to analyze for internal links",
|
|
"minimum": 1,
|
|
}
|
|
},
|
|
"required": ["post_id"],
|
|
},
|
|
"scope": "read",
|
|
},
|
|
]
|
|
|
|
|
|
class PostsHandler:
|
|
"""Handle post-related operations for WordPress"""
|
|
|
|
def __init__(self, client: WordPressClient):
|
|
"""
|
|
Initialize posts handler.
|
|
|
|
Args:
|
|
client: WordPress API client instance
|
|
"""
|
|
self.client = client
|
|
|
|
# === POSTS ===
|
|
|
|
async def list_posts(
|
|
self,
|
|
per_page: int = 10,
|
|
page: int = 1,
|
|
status: str = "any",
|
|
search: str | None = None,
|
|
search_terms: list[str] | None = None,
|
|
include_content: bool = False,
|
|
) -> str:
|
|
"""
|
|
List WordPress posts.
|
|
|
|
Args:
|
|
per_page: Number of posts per page (1-100)
|
|
page: Page number
|
|
status: Post status filter (publish, draft, pending, private, any)
|
|
search: Search term to filter posts
|
|
search_terms: Multiple search terms for parallel search with deduplication
|
|
include_content: Include content summary and word count in results
|
|
|
|
Returns:
|
|
JSON string with posts list
|
|
"""
|
|
try:
|
|
params = {
|
|
"per_page": per_page,
|
|
"page": page,
|
|
"status": status,
|
|
"_embed": "true", # Include author and featured image
|
|
}
|
|
|
|
# Multi-search: parallel API calls with deduplication
|
|
if search_terms and len(search_terms) > 0:
|
|
|
|
async def _search_single(term: str) -> list:
|
|
p = {**params, "search": term}
|
|
return await self.client.get("posts", params=p)
|
|
|
|
batches = await asyncio.gather(
|
|
*[_search_single(term) for term in search_terms], return_exceptions=True
|
|
)
|
|
seen_ids: set = set()
|
|
posts: list = []
|
|
for batch in batches:
|
|
if isinstance(batch, Exception):
|
|
continue
|
|
for post in batch:
|
|
if post["id"] not in seen_ids:
|
|
seen_ids.add(post["id"])
|
|
posts.append(post)
|
|
else:
|
|
if search:
|
|
params["search"] = search
|
|
posts = await self.client.get("posts", params=params)
|
|
|
|
# Format response
|
|
def _format_post(post: dict) -> dict:
|
|
item = {
|
|
"id": post["id"],
|
|
"title": post["title"]["rendered"],
|
|
"excerpt": post["excerpt"]["rendered"][:200],
|
|
"status": post["status"],
|
|
"date": post["date"],
|
|
"author": post.get("_embedded", {})
|
|
.get("author", [{}])[0]
|
|
.get("name", "Unknown"),
|
|
"link": post["link"],
|
|
}
|
|
if include_content:
|
|
content_html = post.get("content", {}).get("rendered", "")
|
|
item["content_summary"] = _strip_html(content_html, 500)
|
|
item["word_count"] = _count_words(content_html)
|
|
return item
|
|
|
|
result = {
|
|
"total": len(posts),
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"posts": [_format_post(post) for post in posts],
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to list posts: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def get_post(self, post_id: int, fields: str | None = None) -> str:
|
|
"""
|
|
Get a specific WordPress post.
|
|
|
|
Args:
|
|
post_id: Post ID to retrieve
|
|
fields: Comma-separated list of fields to return (e.g., 'id,title,status').
|
|
When set, behaves as a STRICT allow-list — only the
|
|
requested names (plus ``id``) are returned. Unknown
|
|
names are ignored.
|
|
|
|
Returns:
|
|
JSON string with post data. Default projection includes
|
|
``featured_media`` (int or 0), ``slug`` (str), and
|
|
``featured_media_url`` (derived from
|
|
``_embedded['wp:featuredmedia'][0].source_url``; empty
|
|
string when no featured image is set) so callers can
|
|
verify a post's featured image was set without a separate
|
|
media-fetch round trip.
|
|
"""
|
|
try:
|
|
# F.X.fix #4: default projection includes the fields every
|
|
# caller asked for during F.X.test (featured_media, slug,
|
|
# featured_media_url). When ``fields`` is supplied we honour
|
|
# it as a STRICT allow-list — callers must opt in to each
|
|
# name explicitly, no hidden always-on subset beyond id.
|
|
params = {"_embed": "true"}
|
|
field_map = {
|
|
"id": "id",
|
|
"title": "title",
|
|
"content": "content",
|
|
"excerpt": "excerpt",
|
|
"status": "status",
|
|
"date": "date",
|
|
"modified": "modified",
|
|
"author": "_embedded",
|
|
"categories": "categories",
|
|
"tags": "tags",
|
|
"link": "link",
|
|
"slug": "slug",
|
|
"featured_media": "featured_media",
|
|
"featured_media_url": "_embedded",
|
|
"word_count": "content",
|
|
}
|
|
if fields:
|
|
wp_fields: set[str] = set()
|
|
requested = {f.strip().lower() for f in fields.split(",") if f.strip()}
|
|
for f in requested:
|
|
if f in field_map:
|
|
wp_fields.add(field_map[f])
|
|
wp_fields.add("id")
|
|
params["_fields"] = ",".join(wp_fields)
|
|
|
|
post = await self.client.get(f"posts/{post_id}", params=params)
|
|
|
|
# Derive featured_media_url from the embedded media entry.
|
|
# WP returns the embedded resource under
|
|
# ``_embedded['wp:featuredmedia']`` (hyphenated key, list of
|
|
# one). Empty string when no featured image is attached.
|
|
featured_media_url = ""
|
|
embedded = post.get("_embedded") or {}
|
|
media_entries = embedded.get("wp:featuredmedia") or []
|
|
if isinstance(media_entries, list) and media_entries:
|
|
first = media_entries[0] or {}
|
|
if isinstance(first, dict):
|
|
featured_media_url = first.get("source_url") or ""
|
|
|
|
# Build full result
|
|
full_result = {
|
|
"id": post["id"],
|
|
"title": post.get("title", {}).get("rendered", ""),
|
|
"slug": post.get("slug", ""),
|
|
"content": post.get("content", {}).get("rendered", ""),
|
|
"excerpt": post.get("excerpt", {}).get("rendered", ""),
|
|
"status": post.get("status", ""),
|
|
"date": post.get("date", ""),
|
|
"modified": post.get("modified", ""),
|
|
"author": embedded.get("author", [{}])[0].get("name", "Unknown"),
|
|
"categories": post.get("categories", []),
|
|
"tags": post.get("tags", []),
|
|
"link": post.get("link", ""),
|
|
"featured_media": post.get("featured_media", 0) or 0,
|
|
"featured_media_url": featured_media_url,
|
|
"word_count": _count_words(post.get("content", {}).get("rendered", "")),
|
|
}
|
|
|
|
# Strict allow-list filtering when ``fields`` is provided.
|
|
if fields:
|
|
requested = {f.strip().lower() for f in fields.split(",") if f.strip()}
|
|
requested.add("id")
|
|
result = {k: v for k, v in full_result.items() if k in requested}
|
|
else:
|
|
result = full_result
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to get post {post_id}: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def create_post(
|
|
self,
|
|
title: str,
|
|
content: str,
|
|
status: str = "draft",
|
|
slug: str | None = None,
|
|
excerpt: str | None = None,
|
|
categories: list[int] | None = None,
|
|
tags: list[int] | None = None,
|
|
featured_media: int | None = None,
|
|
) -> str:
|
|
"""
|
|
Create a new WordPress post.
|
|
|
|
Args:
|
|
title: Post title
|
|
content: Post content (HTML allowed)
|
|
status: Publication status (draft, publish, pending, private, future)
|
|
slug: Post URL slug (auto-generated if not provided)
|
|
excerpt: Post excerpt/summary
|
|
categories: Category IDs to assign
|
|
tags: Tag IDs to assign
|
|
featured_media: Featured image media ID
|
|
|
|
Returns:
|
|
JSON string with created post data
|
|
"""
|
|
try:
|
|
data = {"title": title, "content": content, "status": status}
|
|
if slug:
|
|
data["slug"] = slug
|
|
if excerpt:
|
|
data["excerpt"] = excerpt
|
|
if categories:
|
|
data["categories"] = categories
|
|
if tags:
|
|
data["tags"] = tags
|
|
if featured_media:
|
|
data["featured_media"] = featured_media
|
|
|
|
post = await self.client.post("posts", json_data=data)
|
|
|
|
result = {
|
|
"id": post["id"],
|
|
"title": post["title"]["rendered"],
|
|
"status": post["status"],
|
|
"link": post["link"],
|
|
"message": f"Post created successfully with ID {post['id']}",
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to create post: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def update_post(
|
|
self,
|
|
post_id: int,
|
|
title: str | None = None,
|
|
content: str | None = None,
|
|
status: str | None = None,
|
|
slug: str | None = None,
|
|
excerpt: str | None = None,
|
|
categories: list[int] | None = None,
|
|
tags: list[int] | None = None,
|
|
featured_media: int | None = None,
|
|
) -> str:
|
|
"""
|
|
Update an existing WordPress post.
|
|
|
|
Args:
|
|
post_id: Post ID to update
|
|
title: Post title
|
|
content: Post content
|
|
status: Publication status
|
|
slug: Post URL slug
|
|
excerpt: Post excerpt
|
|
categories: Category IDs
|
|
tags: Tag IDs
|
|
featured_media: Featured image media ID
|
|
|
|
Returns:
|
|
JSON string with updated post data
|
|
"""
|
|
try:
|
|
# Build data dict with only provided values
|
|
data = {}
|
|
if title is not None:
|
|
data["title"] = title
|
|
if content is not None:
|
|
data["content"] = content
|
|
if status is not None:
|
|
data["status"] = status
|
|
if slug is not None:
|
|
data["slug"] = slug
|
|
if excerpt is not None:
|
|
data["excerpt"] = excerpt
|
|
if categories is not None:
|
|
data["categories"] = categories
|
|
if tags is not None:
|
|
data["tags"] = tags
|
|
if featured_media is not None:
|
|
data["featured_media"] = featured_media
|
|
|
|
post = await self.client.post(f"posts/{post_id}", json_data=data)
|
|
|
|
result = {
|
|
"id": post["id"],
|
|
"title": post["title"]["rendered"],
|
|
"status": post["status"],
|
|
"link": post["link"],
|
|
"message": f"Post {post_id} updated successfully",
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to update post {post_id}: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def delete_post(self, post_id: int, force: bool = False) -> str:
|
|
"""
|
|
Delete or trash a WordPress post.
|
|
|
|
Args:
|
|
post_id: Post ID to delete
|
|
force: Permanently delete (True) or move to trash (False)
|
|
|
|
Returns:
|
|
JSON string with deletion result
|
|
"""
|
|
try:
|
|
params = {"force": "true" if force else "false"}
|
|
result = await self.client.delete(f"posts/{post_id}", params=params)
|
|
|
|
message = f"Post {post_id} {'permanently deleted' if force else 'moved to trash'}"
|
|
return json.dumps({"success": True, "message": message, "result": result}, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to delete post {post_id}: {str(e)}"}, indent=2
|
|
)
|
|
|
|
# === PAGES ===
|
|
|
|
async def list_pages(self, per_page: int = 10, page: int = 1, status: str = "any") -> str:
|
|
"""
|
|
List WordPress pages.
|
|
|
|
Args:
|
|
per_page: Number of pages per page (1-100)
|
|
page: Page number
|
|
status: Page status filter
|
|
|
|
Returns:
|
|
JSON string with pages list
|
|
"""
|
|
try:
|
|
params = {"per_page": per_page, "page": page, "status": status}
|
|
pages = await self.client.get("pages", params=params)
|
|
|
|
result = {
|
|
"total": len(pages),
|
|
"pages": [
|
|
{
|
|
"id": p["id"],
|
|
"title": p["title"]["rendered"],
|
|
"status": p["status"],
|
|
"date": p["date"],
|
|
"link": p["link"],
|
|
}
|
|
for p in pages
|
|
],
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to list pages: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def create_page(
|
|
self,
|
|
title: str,
|
|
content: str,
|
|
status: str = "draft",
|
|
slug: str | None = None,
|
|
parent: int | None = None,
|
|
) -> str:
|
|
"""
|
|
Create a new WordPress page.
|
|
|
|
Args:
|
|
title: Page title
|
|
content: Page content (HTML allowed)
|
|
status: Publication status (draft, publish, pending, private)
|
|
slug: Page URL slug (auto-generated if not provided)
|
|
parent: Parent page ID for hierarchical structure
|
|
|
|
Returns:
|
|
JSON string with created page data
|
|
"""
|
|
try:
|
|
data = {"title": title, "content": content, "status": status}
|
|
if slug:
|
|
data["slug"] = slug
|
|
if parent:
|
|
data["parent"] = parent
|
|
|
|
page = await self.client.post("pages", json_data=data)
|
|
|
|
result = {
|
|
"id": page["id"],
|
|
"title": page["title"]["rendered"],
|
|
"status": page["status"],
|
|
"link": page["link"],
|
|
"message": f"Page created successfully with ID {page['id']}",
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to create page: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def update_page(
|
|
self,
|
|
page_id: int,
|
|
title: str | None = None,
|
|
content: str | None = None,
|
|
status: str | None = None,
|
|
slug: str | None = None,
|
|
parent: int | None = None,
|
|
featured_media: int | None = None,
|
|
meta: dict | None = None,
|
|
) -> str:
|
|
"""
|
|
Update an existing WordPress page.
|
|
|
|
Args:
|
|
page_id: Page ID to update
|
|
title: Page title
|
|
content: Page content
|
|
status: Publication status
|
|
slug: Page URL slug
|
|
parent: Parent page ID
|
|
featured_media: Featured image attachment ID (0 clears it)
|
|
meta: Post meta key/value map (e.g. Yoast fields)
|
|
|
|
Returns:
|
|
JSON string with updated page data
|
|
"""
|
|
try:
|
|
# Build data dict with only provided values
|
|
data = {}
|
|
if title is not None:
|
|
data["title"] = title
|
|
if content is not None:
|
|
data["content"] = content
|
|
if status is not None:
|
|
data["status"] = status
|
|
if slug is not None:
|
|
data["slug"] = slug
|
|
if parent is not None:
|
|
data["parent"] = parent
|
|
if featured_media is not None:
|
|
data["featured_media"] = featured_media
|
|
if meta is not None:
|
|
data["meta"] = meta
|
|
|
|
page = await self.client.post(f"pages/{page_id}", json_data=data)
|
|
|
|
result = {
|
|
"id": page["id"],
|
|
"title": page["title"]["rendered"],
|
|
"status": page["status"],
|
|
"link": page["link"],
|
|
"message": f"Page {page_id} updated successfully",
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to update page {page_id}: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def delete_page(self, page_id: int, force: bool = False) -> str:
|
|
"""
|
|
Delete or trash a WordPress page.
|
|
|
|
Args:
|
|
page_id: Page ID to delete
|
|
force: Permanently delete (True) or move to trash (False)
|
|
|
|
Returns:
|
|
JSON string with deletion result
|
|
"""
|
|
try:
|
|
# Phase K.2.4: Convert boolean to string for WP REST API
|
|
params = {"force": "true" if force else "false"}
|
|
await self.client.delete(f"pages/{page_id}", params=params)
|
|
|
|
if force:
|
|
message = f"Page {page_id} permanently deleted"
|
|
else:
|
|
message = f"Page {page_id} moved to trash"
|
|
|
|
result = {"id": page_id, "deleted": True, "force": force, "message": message}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to delete page {page_id}: {str(e)}"}, indent=2
|
|
)
|
|
|
|
# === POST TYPES ===
|
|
|
|
async def list_post_types(self) -> str:
|
|
"""
|
|
List all registered post types.
|
|
|
|
Returns list of all post types including built-in (post, page)
|
|
and custom post types (portfolio, testimonials, etc.).
|
|
|
|
Returns:
|
|
JSON string with total count and post types list
|
|
"""
|
|
try:
|
|
types_data = await self.client.get("types")
|
|
|
|
# Convert dict to list
|
|
post_types = []
|
|
if isinstance(types_data, dict):
|
|
for slug, data in types_data.items():
|
|
post_types.append(
|
|
{
|
|
"slug": slug,
|
|
"name": data.get("name", slug),
|
|
"description": data.get("description", ""),
|
|
"rest_base": data.get("rest_base", slug),
|
|
"hierarchical": data.get("hierarchical", False),
|
|
"supports": data.get("supports", {}),
|
|
}
|
|
)
|
|
|
|
result = {"total": len(post_types), "post_types": post_types}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{"error": str(e), "message": f"Failed to list post types: {str(e)}"}, indent=2
|
|
)
|
|
|
|
async def get_post_type_info(self, post_type: str) -> str:
|
|
"""
|
|
Get detailed information about a specific post type.
|
|
|
|
Args:
|
|
post_type: Post type slug (e.g., 'portfolio', 'post', 'page')
|
|
|
|
Returns:
|
|
JSON string with post type details
|
|
"""
|
|
try:
|
|
info = await self.client.get(f"types/{post_type}")
|
|
return json.dumps(info, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{
|
|
"error": str(e),
|
|
"message": f"Failed to get post type info for '{post_type}': {str(e)}",
|
|
},
|
|
indent=2,
|
|
)
|
|
|
|
# === CUSTOM POSTS ===
|
|
|
|
async def list_custom_posts(
|
|
self, post_type: str, per_page: int = 10, page: int = 1, status: str = "any"
|
|
) -> str:
|
|
"""
|
|
List posts of a specific custom post type.
|
|
|
|
Args:
|
|
post_type: Post type slug (e.g., 'portfolio', 'testimonials')
|
|
per_page: Number of posts per page
|
|
page: Page number
|
|
status: Post status filter
|
|
|
|
Returns:
|
|
JSON string with posts list
|
|
"""
|
|
try:
|
|
params = {"per_page": per_page, "page": page, "status": status, "_embed": "true"}
|
|
|
|
# Use the post type's rest_base as endpoint
|
|
posts = await self.client.get(post_type, params=params)
|
|
|
|
result = {
|
|
"total": len(posts) if isinstance(posts, list) else 0,
|
|
"post_type": post_type,
|
|
"page": page,
|
|
"per_page": per_page,
|
|
"posts": posts,
|
|
}
|
|
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{
|
|
"error": str(e),
|
|
"message": f"Failed to list custom posts of type '{post_type}': {str(e)}",
|
|
},
|
|
indent=2,
|
|
)
|
|
|
|
async def create_custom_post(
|
|
self,
|
|
post_type: str,
|
|
title: str,
|
|
content: str,
|
|
status: str = "draft",
|
|
meta: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""
|
|
Create a new post of a custom post type.
|
|
|
|
Args:
|
|
post_type: Post type slug (e.g., 'portfolio')
|
|
title: Post title
|
|
content: Post content (HTML allowed)
|
|
status: Post status (draft, publish, etc.)
|
|
meta: Custom fields/meta data
|
|
|
|
Returns:
|
|
JSON string with created post details
|
|
"""
|
|
try:
|
|
data = {"title": title, "content": content, "status": status}
|
|
|
|
if meta:
|
|
data["meta"] = meta
|
|
|
|
post = await self.client.post(post_type, json_data=data)
|
|
|
|
return json.dumps(post, indent=2)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{
|
|
"error": str(e),
|
|
"message": f"Failed to create custom post of type '{post_type}': {str(e)}",
|
|
},
|
|
indent=2,
|
|
)
|
|
|
|
# === INTERNAL LINKS ===
|
|
|
|
async def get_internal_links(self, post_id: int) -> str:
|
|
"""
|
|
Extract internal links from a post's content.
|
|
|
|
Args:
|
|
post_id: Post ID to analyze
|
|
|
|
Returns:
|
|
JSON string with internal and external link counts
|
|
"""
|
|
try:
|
|
post = await self.client.get(
|
|
f"posts/{post_id}", params={"_fields": "id,title,content,link"}
|
|
)
|
|
|
|
content_html = post.get("content", {}).get("rendered", "")
|
|
post_title = post.get("title", {}).get("rendered", "")
|
|
post_link = post.get("link", "")
|
|
|
|
# Extract site URL from post link
|
|
from urllib.parse import urlparse
|
|
|
|
parsed = urlparse(post_link)
|
|
site_url = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme else ""
|
|
|
|
# Extract all <a> tags with href and anchor text
|
|
link_pattern = re.compile(
|
|
r'<a\s[^>]*href=["\']([^"\']+)["\'][^>]*>(.*?)</a>',
|
|
re.IGNORECASE | re.DOTALL,
|
|
)
|
|
matches = link_pattern.findall(content_html)
|
|
|
|
internal_links = []
|
|
external_count = 0
|
|
|
|
for href, anchor_html in matches:
|
|
anchor_text = re.sub(r"<[^>]+>", "", anchor_html).strip()
|
|
href = href.strip()
|
|
|
|
# Skip anchors and empty hrefs
|
|
if not href or href.startswith("#") or href.startswith("mailto:"):
|
|
continue
|
|
|
|
# Check if internal
|
|
is_internal = False
|
|
if site_url and href.startswith(site_url):
|
|
is_internal = True
|
|
elif href.startswith("/") and not href.startswith("//"):
|
|
is_internal = True
|
|
href = site_url + href
|
|
|
|
if is_internal:
|
|
internal_links.append(
|
|
{
|
|
"url": href,
|
|
"anchor_text": anchor_text,
|
|
}
|
|
)
|
|
else:
|
|
external_count += 1
|
|
|
|
result = {
|
|
"post_id": post_id,
|
|
"post_title": post_title,
|
|
"site_url": site_url,
|
|
"total_internal_links": len(internal_links),
|
|
"internal_links": internal_links,
|
|
"external_links_count": external_count,
|
|
}
|
|
|
|
return json.dumps(result, indent=2, ensure_ascii=False)
|
|
except Exception as e:
|
|
return json.dumps(
|
|
{
|
|
"error": str(e),
|
|
"message": f"Failed to get internal links for post {post_id}: {str(e)}",
|
|
},
|
|
indent=2,
|
|
)
|