"""
System Operations Handler
Manages WordPress system operations including:
- System information (PHP, MySQL, WordPress versions)
- Disk usage statistics
- Cron job management
- Cache operations
- Error log retrieval
Most operations require WP-CLI for advanced functionality.
"""
import json
import re
from typing import Any
from plugins.wordpress.client import WordPressClient
from plugins.wordpress.wp_cli import WPCLIManager
def get_tool_specifications() -> list[dict[str, Any]]:
"""Return tool specifications for ToolGenerator"""
return [
# System Info
{
"name": "system_info",
"method_name": "system_info",
"description": "Get comprehensive system information including PHP version, MySQL version, WordPress version, server info, memory limits, and more.",
"schema": {"type": "object", "properties": {}},
"scope": "read",
},
# PHP Info
{
"name": "system_phpinfo",
"method_name": "system_phpinfo",
"description": "Get detailed PHP configuration including loaded extensions, ini settings, and disabled functions.",
"schema": {"type": "object", "properties": {}},
"scope": "read",
},
# Disk Usage
{
"name": "system_disk_usage",
"method_name": "system_disk_usage",
"description": "Get disk usage statistics including uploads size, plugins size, themes size, and database size.",
"schema": {"type": "object", "properties": {}},
"scope": "read",
},
# Clear All Caches
{
"name": "system_clear_all_caches",
"method_name": "system_clear_all_caches",
"description": "Clear all caches including object cache, transients, and opcache (if available). Safe to run anytime.",
"schema": {"type": "object", "properties": {}},
"scope": "write",
},
# Cron List
{
"name": "cron_list",
"method_name": "cron_list",
"description": "List all scheduled WordPress cron jobs with schedule, next run time, and arguments.",
"schema": {"type": "object", "properties": {}},
"scope": "read",
},
# Cron Run
{
"name": "cron_run",
"method_name": "cron_run",
"description": "Manually trigger a specific cron job by hook name. Useful for testing or forcing scheduled tasks.",
"schema": {
"type": "object",
"properties": {
"hook": {"type": "string", "description": "Cron hook name to execute"},
"args": {
"type": "array",
"items": {},
"default": [],
"description": "Optional arguments to pass to the hook",
},
},
"required": ["hook"],
},
"scope": "write",
},
# Error Log
{
"name": "error_log",
"method_name": "error_log",
"description": "Get recent PHP error log entries. Useful for debugging issues. Admin scope recommended for security.",
"schema": {
"type": "object",
"properties": {
"lines": {
"type": "integer",
"default": 100,
"minimum": 1,
"maximum": 1000,
"description": "Number of log lines to retrieve",
},
"filter": {
"type": "string",
"description": "Filter logs by keyword (case-insensitive)",
},
"level": {
"type": "string",
"enum": ["error", "warning", "notice", "fatal"],
"description": "Filter by error level",
},
},
},
"scope": "read",
},
]
class SystemHandler:
"""Handles WordPress system operations"""
def __init__(self, client: WordPressClient, wp_cli: WPCLIManager | None = None):
"""
Initialize System Handler
Args:
client: WordPress REST API client
wp_cli: WP-CLI manager (optional, for advanced operations)
"""
self.client = client
self.wp_cli = wp_cli
self.logger = client.logger
async def wp_cli_version(self) -> dict[str, Any]:
"""Get WP-CLI version information"""
if not self.wp_cli:
return {"success": False, "version": None, "error": "WP-CLI not available"}
try:
result = await self.wp_cli.execute_command("cli version")
version_output = result.get("output", "").strip()
# Parse version (e.g., "WP-CLI 2.10.0")
version = (
version_output.replace("WP-CLI ", "").strip()
if "WP-CLI" in version_output
else version_output
)
return {"success": True, "version": version, "full_output": version_output}
except Exception as e:
self.logger.error(f"WP-CLI version check failed: {str(e)}")
return {"success": False, "version": None, "error": str(e)}
async def system_info(self) -> dict[str, Any]:
"""Get comprehensive system information"""
if not self.wp_cli:
return {
"success": False,
"error": "WP-CLI not available. Container name not configured.",
}
try:
# Get various system info using WP-CLI
info_commands = {
"php_version": "eval 'echo PHP_VERSION;'",
"wordpress_version": "core version",
"site_url": "option get siteurl",
"active_theme": "theme list --status=active --field=name",
"plugin_count": "plugin list --status=active --format=count",
"wp_debug": "config get WP_DEBUG",
"wp_debug_log": "config get WP_DEBUG_LOG",
"multisite": "config get MULTISITE",
}
results = {}
for key, cmd in info_commands.items():
try:
result = await self.wp_cli.execute_command(cmd)
results[key] = result.get("output", "").strip()
except Exception as e:
self.logger.warning(f"Failed to get {key}: {str(e)}")
results[key] = "N/A"
# Get PHP settings
php_settings_cmd = """eval 'echo json_encode([
"memory_limit" => ini_get("memory_limit"),
"max_execution_time" => ini_get("max_execution_time"),
"upload_max_filesize" => ini_get("upload_max_filesize"),
"post_max_size" => ini_get("post_max_size"),
"max_input_vars" => ini_get("max_input_vars")
]);'"""
php_result = await self.wp_cli.execute_command(php_settings_cmd)
php_settings = json.loads(php_result.get("output", "{}"))
# Get MySQL version
mysql_cmd = 'db query "SELECT VERSION()" --skip-column-names'
mysql_result = await self.wp_cli.execute_command(mysql_cmd)
mysql_version = mysql_result.get("output", "N/A").strip()
# Get server software from environment
server_cmd = 'eval \'echo $_SERVER["SERVER_SOFTWARE"] ?? "Unknown";\''
server_result = await self.wp_cli.execute_command(server_cmd)
server_software = server_result.get("output", "Unknown").strip()
# Get loaded PHP extensions
ext_cmd = "eval 'echo json_encode(get_loaded_extensions());'"
ext_result = await self.wp_cli.execute_command(ext_cmd)
php_extensions = json.loads(ext_result.get("output", "[]"))
return {
"success": True,
"php_version": results.get("php_version", "N/A"),
"mysql_version": mysql_version,
"wordpress_version": results.get("wordpress_version", "N/A"),
"server_software": server_software,
"memory_limit": php_settings.get("memory_limit", "N/A"),
"max_execution_time": int(php_settings.get("max_execution_time", 0)),
"upload_max_filesize": php_settings.get("upload_max_filesize", "N/A"),
"post_max_size": php_settings.get("post_max_size", "N/A"),
"max_input_vars": int(php_settings.get("max_input_vars", 0)),
"php_extensions": php_extensions,
"wp_debug": results.get("wp_debug", "false") == "true",
"wp_debug_log": results.get("wp_debug_log", "false") == "true",
"multisite": results.get("multisite", "false") == "true",
"active_plugins": int(results.get("plugin_count", 0)),
"active_theme": results.get("active_theme", "N/A"),
"site_url": results.get("site_url", "N/A"),
}
except Exception as e:
self.logger.error(f"System info failed: {str(e)}")
return {"success": False, "error": str(e)}
async def system_phpinfo(self) -> dict[str, Any]:
"""Get detailed PHP configuration"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
# Get PHP version and SAPI
version_cmd = "eval 'echo PHP_VERSION;'"
version_result = await self.wp_cli.execute_command(version_cmd)
php_version = version_result.get("output", "").strip()
sapi_cmd = "eval 'echo PHP_SAPI;'"
sapi_result = await self.wp_cli.execute_command(sapi_cmd)
php_sapi = sapi_result.get("output", "").strip()
# Get loaded extensions
ext_cmd = "eval 'echo json_encode(get_loaded_extensions());'"
ext_result = await self.wp_cli.execute_command(ext_cmd)
extensions = json.loads(ext_result.get("output", "[]"))
# Get important ini settings
ini_settings_cmd = """eval 'echo json_encode([
"display_errors" => ini_get("display_errors"),
"error_reporting" => ini_get("error_reporting"),
"log_errors" => ini_get("log_errors"),
"error_log" => ini_get("error_log"),
"memory_limit" => ini_get("memory_limit"),
"max_execution_time" => ini_get("max_execution_time"),
"upload_max_filesize" => ini_get("upload_max_filesize"),
"post_max_size" => ini_get("post_max_size"),
"max_input_vars" => ini_get("max_input_vars"),
"max_input_time" => ini_get("max_input_time"),
"default_socket_timeout" => ini_get("default_socket_timeout"),
"allow_url_fopen" => ini_get("allow_url_fopen"),
"session.save_handler" => ini_get("session.save_handler")
]);'"""
ini_result = await self.wp_cli.execute_command(ini_settings_cmd)
ini_settings = json.loads(ini_result.get("output", "{}"))
# Get disabled functions
disabled_cmd = "eval 'echo ini_get(\"disable_functions\");'"
disabled_result = await self.wp_cli.execute_command(disabled_cmd)
disabled_raw = disabled_result.get("output", "").strip()
disabled_functions = [f.strip() for f in disabled_raw.split(",") if f.strip()]
return {
"success": True,
"version": php_version,
"sapi": php_sapi,
"extensions": extensions,
"ini_settings": ini_settings,
"disabled_functions": disabled_functions,
}
except Exception as e:
self.logger.error(f"PHP info failed: {str(e)}")
return {"success": False, "error": str(e)}
async def system_disk_usage(self) -> dict[str, Any]:
"""Get disk usage statistics"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
# Get WordPress root path
path_cmd = "eval 'echo ABSPATH;'"
path_result = await self.wp_cli.execute_command(path_cmd)
wp_path = path_result.get("output", "").strip()
# Get directory sizes using du command
sizes = {}
# Uploads directory
uploads_cmd = f"eval 'echo shell_exec(\"du -sm {wp_path}wp-content/uploads 2>/dev/null | cut -f1\");'"
try:
uploads_result = await self.wp_cli.execute_command(uploads_cmd)
upload_size = uploads_result.get("output", "0").strip()
sizes["uploads_size_mb"] = (
float(upload_size) if upload_size and upload_size != "" else 0.0
)
except Exception as e:
self.logger.warning(f"Failed to get uploads size: {e}")
sizes["uploads_size_mb"] = 0.0
# Plugins directory
plugins_cmd = f"eval 'echo shell_exec(\"du -sm {wp_path}wp-content/plugins 2>/dev/null | cut -f1\");'"
try:
plugins_result = await self.wp_cli.execute_command(plugins_cmd)
plugins_size = plugins_result.get("output", "0").strip()
sizes["plugins_size_mb"] = (
float(plugins_size) if plugins_size and plugins_size != "" else 0.0
)
except Exception as e:
self.logger.warning(f"Failed to get plugins size: {e}")
sizes["plugins_size_mb"] = 0.0
# Themes directory
themes_cmd = f"eval 'echo shell_exec(\"du -sm {wp_path}wp-content/themes 2>/dev/null | cut -f1\");'"
try:
themes_result = await self.wp_cli.execute_command(themes_cmd)
themes_size = themes_result.get("output", "0").strip()
sizes["themes_size_mb"] = (
float(themes_size) if themes_size and themes_size != "" else 0.0
)
except Exception as e:
self.logger.warning(f"Failed to get themes size: {e}")
sizes["themes_size_mb"] = 0.0
# Total WordPress directory
total_cmd = f"eval 'echo shell_exec(\"du -sm {wp_path} 2>/dev/null | cut -f1\");'"
try:
total_result = await self.wp_cli.execute_command(total_cmd)
total_size = total_result.get("output", "0").strip()
sizes["wordpress_size_mb"] = (
float(total_size) if total_size and total_size != "" else 0.0
)
except Exception as e:
self.logger.warning(f"Failed to get wordpress total size: {e}")
sizes["wordpress_size_mb"] = 0.0
# Database size (from our database handler logic)
db_query = """
SELECT ROUND(SUM(data_length + index_length) / 1024 / 1024, 2) AS size_mb
FROM information_schema.TABLES
WHERE table_schema = DATABASE()
"""
db_cmd = f'db query "{db_query}" --skip-column-names'
try:
db_result = await self.wp_cli.execute_command(db_cmd)
sizes["database_size_mb"] = float(db_result.get("output", "0").strip())
except (ValueError, KeyError, Exception):
sizes["database_size_mb"] = 0.0
# Calculate total
total_size = sum([sizes["wordpress_size_mb"], sizes["database_size_mb"]])
return {
"success": True,
"total_size_mb": round(total_size, 2),
**sizes,
"breakdown": {
"uploads": sizes["uploads_size_mb"],
"plugins": sizes["plugins_size_mb"],
"themes": sizes["themes_size_mb"],
"database": sizes["database_size_mb"],
},
}
except Exception as e:
self.logger.error(f"Disk usage check failed: {str(e)}")
return {"success": False, "error": str(e)}
async def system_clear_all_caches(self) -> dict[str, Any]:
"""Clear all caches"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
cleared = []
# Clear object cache
try:
await self.wp_cli.execute_command("cache flush")
cleared.append("object_cache")
except Exception as e:
self.logger.warning(f"Object cache flush failed: {str(e)}")
# Clear transients
try:
await self.wp_cli.execute_command("transient delete --all")
cleared.append("transients")
except Exception as e:
self.logger.warning(f"Transient delete failed: {str(e)}")
# Clear OPcache if available
try:
opcache_cmd = 'eval \'if (function_exists("opcache_reset")) { opcache_reset(); echo "cleared"; }\''
opcache_result = await self.wp_cli.execute_command(opcache_cmd)
if "cleared" in opcache_result.get("output", ""):
cleared.append("opcache")
except Exception as e:
self.logger.warning(f"OPcache clear failed: {str(e)}")
return {
"success": True,
"message": f"Cleared {len(cleared)} cache types",
"cleared": cleared,
}
except Exception as e:
self.logger.error(f"Clear all caches failed: {str(e)}")
return {"success": False, "error": str(e)}
async def cron_list(self) -> dict[str, Any]:
"""List all WordPress cron jobs"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
# Get cron events
cmd = "cron event list --format=json"
result = await self.wp_cli.execute_command(cmd)
events_data = json.loads(result.get("output", "[]"))
# Parse events
events = []
for event in events_data:
events.append(
{
"hook": event.get("hook", ""),
"timestamp": int(event.get("time", 0)),
"schedule": event.get("recurrence", "single"),
"interval": event.get("interval"),
"args": event.get("args", []),
}
)
# Get available schedules
schedules_cmd = "cron schedule list --format=json"
try:
schedules_result = await self.wp_cli.execute_command(schedules_cmd)
schedules_data = json.loads(schedules_result.get("output", "[]"))
schedules = {s.get("name"): s for s in schedules_data}
except (json.JSONDecodeError, KeyError, Exception):
schedules = {}
return {"success": True, "events": events, "total": len(events), "schedules": schedules}
except Exception as e:
self.logger.error(f"Cron list failed: {str(e)}")
return {"success": False, "error": str(e)}
async def cron_run(self, hook: str, args: list[Any] | None = None) -> dict[str, Any]:
"""Manually run a cron job"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
# Build command
cmd = f"cron event run {hook}"
# Execute cron job
result = await self.wp_cli.execute_command(cmd)
return {
"success": True,
"message": f"Cron job '{hook}' executed",
"hook": hook,
"output": result.get("output", ""),
}
except Exception as e:
self.logger.error(f"Cron run failed: {str(e)}")
return {"success": False, "error": str(e)}
async def error_log(
self, lines: int = 100, filter: str | None = None, level: str | None = None
) -> dict[str, Any]:
"""Get PHP error log entries"""
if not self.wp_cli:
return {"success": False, "error": "WP-CLI not available"}
try:
# Get error log path
log_path_cmd = "eval 'echo ini_get(\"error_log\");'"
log_path_result = await self.wp_cli.execute_command(log_path_cmd)
log_path = log_path_result.get("output", "").strip()
if not log_path or log_path == "":
# Try WordPress debug log
wp_path_cmd = "eval 'echo WP_CONTENT_DIR;'"
wp_path_result = await self.wp_cli.execute_command(wp_path_cmd)
wp_content = wp_path_result.get("output", "").strip()
log_path = f"{wp_content}/debug.log"
# Get log file size
size_cmd = f"eval 'echo filesize(\"{log_path}\") ?? 0;'"
try:
size_result = await self.wp_cli.execute_command(size_cmd)
log_size_bytes = int(size_result.get("output", "0").strip())
log_size_mb = round(log_size_bytes / 1024 / 1024, 2)
except (ValueError, KeyError, Exception):
log_size_mb = 0.0
# Read log file (last N lines)
tail_cmd = f"eval 'exec(\"tail -n {lines} {log_path} 2>&1\");'"
log_result = await self.wp_cli.execute_command(tail_cmd)
log_lines = log_result.get("output", "").strip().split("\n")
# Parse log entries
entries = []
for line in log_lines:
if not line.strip():
continue
# Basic parsing (PHP error log format varies)
# Format: [timestamp] PHP Error_Type: message in file on line X
match = re.match(r"\[([^\]]+)\]\s+PHP\s+(\w+):\s+(.+)", line)
if match:
timestamp, error_type, message = match.groups()
# Extract file and line if present
file_match = re.search(r"in\s+(.+)\s+on\s+line\s+(\d+)", message)
file_path = file_match.group(1) if file_match else None
line_num = int(file_match.group(2)) if file_match else None
entry = {
"timestamp": timestamp,
"level": error_type.lower(),
"message": message,
"file": file_path,
"line": line_num,
}
# Apply filters
if level and entry["level"] != level.lower():
continue
if filter and filter.lower() not in message.lower():
continue
entries.append(entry)
else:
# Unparsed line - include as-is
entries.append(
{
"timestamp": "N/A",
"level": "unknown",
"message": line,
"file": None,
"line": None,
}
)
return {
"success": True,
"entries": entries,
"total_lines": len(log_lines),
"filtered_lines": len(entries),
"log_size_mb": log_size_mb,
"log_path": log_path,
}
except Exception as e:
self.logger.error(f"Error log retrieval failed: {str(e)}")
return {"success": False, "error": str(e)}
