""" 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 REST API access (primary health indicator — most tools use REST) rest_api_available = False try: 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: {e}") # Test WP-CLI access (optional — only needed for database/system tools) wp_cli_available = False try: wp_cli_version = await self.system.wp_cli_version() wp_cli_available = bool(wp_cli_version.get("version")) except Exception: pass return { "healthy": rest_api_available, "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