c73e39f6e4
- Fix Python formatting (sync no longer strips blank lines from .py files) - Remove test_community_build.py (tests private sync module) - Fix ruff warnings in test files - Add CODE_OF_CONDUCT.md - Add .github/PULL_REQUEST_TEMPLATE.md Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""
|
|
WordPress Advanced Plugin - Clean Architecture
|
|
|
|
Advanced WordPress management features requiring elevated permissions.
|
|
Provides database operations, bulk operations, and system management.
|
|
|
|
This plugin is separated from core WordPress plugin for:
|
|
- Better security (separate API keys for advanced features)
|
|
- Better tool visibility (users see only features they need)
|
|
- Granular access control
|
|
"""
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from plugins.base import BasePlugin
|
|
from plugins.wordpress.client import WordPressClient
|
|
from plugins.wordpress_advanced import handlers
|
|
|
|
|
|
class WordPressAdvancedPlugin(BasePlugin):
|
|
"""
|
|
WordPress Advanced plugin - separated for security and visibility.
|
|
|
|
Provides advanced WordPress management capabilities:
|
|
- Database operations (export, import, search, query, repair)
|
|
- Bulk operations (batch updates/deletes for posts, products, media)
|
|
- System operations (system info, cache, cron, error logs)
|
|
|
|
Requires:
|
|
- WordPress site with REST API
|
|
- WP-CLI access (for database and system operations)
|
|
- Docker container name (for WP-CLI execution)
|
|
"""
|
|
|
|
@staticmethod
|
|
def get_plugin_name() -> str:
|
|
"""Return plugin type identifier"""
|
|
return "wordpress_advanced"
|
|
|
|
@staticmethod
|
|
def get_required_config_keys() -> list[str]:
|
|
"""Return required configuration keys"""
|
|
return ["url", "username", "app_password", "container"]
|
|
|
|
def __init__(self, config: dict[str, Any], project_id: str | None = None):
|
|
"""
|
|
Initialize WordPress Advanced plugin with handlers.
|
|
|
|
Args:
|
|
config: Configuration dictionary containing:
|
|
- url: WordPress site URL
|
|
- username: WordPress username
|
|
- app_password: WordPress application password
|
|
- container: Docker container name for WP-CLI (REQUIRED)
|
|
project_id: Optional project ID (auto-generated if not provided)
|
|
"""
|
|
super().__init__(config, project_id=project_id)
|
|
|
|
# Create WordPress API client
|
|
self.client = WordPressClient(
|
|
site_url=config["url"], username=config["username"], app_password=config["app_password"]
|
|
)
|
|
|
|
# WP-CLI is REQUIRED for wordpress_advanced
|
|
container_name = config.get("container")
|
|
if not container_name:
|
|
raise ValueError(
|
|
"WordPress Advanced plugin requires 'container' configuration. "
|
|
"Please set WORDPRESS_ADVANCED_SITE1_CONTAINER in environment variables."
|
|
)
|
|
|
|
# Import WP-CLI manager
|
|
from plugins.wordpress.wp_cli import WPCLIManager
|
|
|
|
wp_cli_manager = WPCLIManager(container_name)
|
|
|
|
# Initialize handlers (all require WP-CLI or advanced REST API)
|
|
self.database = handlers.DatabaseHandler(self.client, wp_cli_manager)
|
|
self.bulk = handlers.BulkHandler(self.client)
|
|
self.system = handlers.SystemHandler(self.client, wp_cli_manager)
|
|
|
|
@staticmethod
|
|
def get_tool_specifications() -> list[dict[str, Any]]:
|
|
"""
|
|
Return all tool specifications for ToolGenerator.
|
|
|
|
This method is called by ToolGenerator to create unified tools
|
|
with site parameter routing.
|
|
|
|
Returns:
|
|
List of tool specification dictionaries (22 tools total)
|
|
"""
|
|
specs = []
|
|
|
|
# Database operations (7 tools)
|
|
specs.extend(handlers.get_database_specs())
|
|
|
|
# Bulk operations (8 tools)
|
|
specs.extend(handlers.get_bulk_specs())
|
|
|
|
# System operations (7 tools)
|
|
specs.extend(handlers.get_system_specs())
|
|
|
|
return specs
|
|
|
|
async def health_check(self) -> dict[str, Any]:
|
|
"""
|
|
Check WordPress Advanced features availability.
|
|
|
|
Returns:
|
|
Dict with health status and WP-CLI availability
|
|
"""
|
|
try:
|
|
# Test WP-CLI access (primary requirement for wordpress_advanced)
|
|
wp_cli_version = await self.system.wp_cli_version()
|
|
wp_cli_available = bool(wp_cli_version.get("version"))
|
|
|
|
# Test REST API access with a public endpoint
|
|
rest_api_available = False
|
|
try:
|
|
# Use a public endpoint that doesn't require authentication
|
|
site_info = await self.client.get("/")
|
|
rest_api_available = bool(site_info.get("name"))
|
|
except Exception as e:
|
|
self.logger.warning(f"REST API check failed (non-critical): {e}")
|
|
rest_api_available = False
|
|
|
|
return {
|
|
"healthy": wp_cli_available, # Only WP-CLI is critical for wordpress_advanced
|
|
"wp_cli_available": wp_cli_available,
|
|
"rest_api_available": rest_api_available,
|
|
"features": {
|
|
"database_operations": wp_cli_available,
|
|
"bulk_operations": rest_api_available,
|
|
"system_operations": wp_cli_available,
|
|
},
|
|
}
|
|
except Exception as e:
|
|
return {
|
|
"healthy": False,
|
|
"error": str(e),
|
|
"wp_cli_available": False,
|
|
"rest_api_available": False,
|
|
}
|
|
|
|
# ========================================
|
|
# Method Delegation to Handlers
|
|
# ========================================
|
|
# All methods delegate to appropriate handlers
|
|
# This maintains backward compatibility with existing code
|
|
|
|
# === Database Operations (7 tools) ===
|
|
async def wp_db_export(self, **kwargs):
|
|
"""Export WordPress database"""
|
|
result = await self.database.wp_db_export(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_import(self, **kwargs):
|
|
"""Import WordPress database"""
|
|
result = await self.database.wp_db_import(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_size(self, **kwargs):
|
|
"""Get database size"""
|
|
result = await self.database.wp_db_size(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_tables(self, **kwargs):
|
|
"""List database tables"""
|
|
result = await self.database.wp_db_tables(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_search(self, **kwargs):
|
|
"""Search database"""
|
|
result = await self.database.wp_db_search(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_query(self, **kwargs):
|
|
"""Execute read-only database query"""
|
|
result = await self.database.wp_db_query(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def wp_db_repair(self, **kwargs):
|
|
"""Repair and optimize database"""
|
|
result = await self.database.wp_db_repair(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
# === Bulk Operations (8 tools) ===
|
|
async def bulk_update_posts(self, **kwargs):
|
|
"""Bulk update posts"""
|
|
result = await self.bulk.bulk_update_posts(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_delete_posts(self, **kwargs):
|
|
"""Bulk delete posts"""
|
|
result = await self.bulk.bulk_delete_posts(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_update_products(self, **kwargs):
|
|
"""Bulk update products"""
|
|
result = await self.bulk.bulk_update_products(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_delete_products(self, **kwargs):
|
|
"""Bulk delete products"""
|
|
result = await self.bulk.bulk_delete_products(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_delete_media(self, **kwargs):
|
|
"""Bulk delete media"""
|
|
result = await self.bulk.bulk_delete_media(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_assign_categories(self, **kwargs):
|
|
"""Bulk assign categories to posts"""
|
|
result = await self.bulk.bulk_assign_categories(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_assign_tags(self, **kwargs):
|
|
"""Bulk assign tags to posts"""
|
|
result = await self.bulk.bulk_assign_tags(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def bulk_trash_posts(self, **kwargs):
|
|
"""Bulk move posts to trash"""
|
|
result = await self.bulk.bulk_trash_posts(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
# === System Operations (7 tools) ===
|
|
async def system_info(self, **kwargs):
|
|
"""Get system information"""
|
|
result = await self.system.system_info(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def system_phpinfo(self, **kwargs):
|
|
"""Get PHP information"""
|
|
result = await self.system.system_phpinfo(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def system_disk_usage(self, **kwargs):
|
|
"""Get disk usage"""
|
|
result = await self.system.system_disk_usage(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def system_clear_all_caches(self, **kwargs):
|
|
"""Clear all caches"""
|
|
result = await self.system.system_clear_all_caches(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def cron_list(self, **kwargs):
|
|
"""List cron jobs"""
|
|
result = await self.system.cron_list(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def cron_run(self, **kwargs):
|
|
"""Run cron job"""
|
|
result = await self.system.cron_run(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|
|
|
|
async def error_log(self, **kwargs):
|
|
"""Get error log"""
|
|
result = await self.system.error_log(**kwargs)
|
|
return json.dumps(result, indent=2) if isinstance(result, dict) else result
|