""" Operation Parser Service - Phase 16 R4 瘦身 Router 抽取 ======================================================= 從 approvals.py 抽取操作解析邏輯，支援中英文指令格式。 功能: - 從 action 字串解析 OperationType - 提取資源名稱與 Namespace - 支援 kubectl 指令與中文自然語言 版本: v1.0 建立: 2026-03-25 (台北時區) 建立者: Claude Code (Phase 16 R4) """ import re from dataclasses import dataclass from src.services.executor import OperationType @dataclass class ParsedOperation: """ 解析後的操作資訊 Attributes: operation_type: K8s 操作類型 (RESTART_DEPLOYMENT, DELETE_POD, etc.) resource_name: 目標資源名稱 namespace: K8s Namespace (預設 "default") Note: 支援 tuple 解包以向後兼容: op, resource, ns = parse_operation_from_action(action) """ operation_type: OperationType | None resource_name: str | None namespace: str def __iter__(self): """支援 tuple 解包: op, resource, ns = parsed""" return iter((self.operation_type, self.resource_name, self.namespace)) def parse_operation_from_action(action: str) -> ParsedOperation: """ 從 action 字串解析操作類型與目標資源 支援格式: - kubectl 指令: "kubectl delete pod nginx-xxx -n production" - 英文自然語言: "Restart deployment api-backend" - 中文自然語言: "重新啟動 awoooi-worker 服務" Examples: "kubectl delete pod nginx-xxx -n production" → ParsedOperation(DELETE_POD, "nginx-xxx", "production") "Restart deployment api-backend" → ParsedOperation(RESTART_DEPLOYMENT, "api-backend", "default") "Scale deployment web-frontend to 5 replicas" → ParsedOperation(SCALE_DEPLOYMENT, "web-frontend", "default") "重新啟動 awoooi-worker 服務" → ParsedOperation(RESTART_DEPLOYMENT, "awoooi-worker", "default") Args: action: 操作指令字串 Returns: ParsedOperation: 解析後的操作資訊 """ action_lower = action.lower() # Pattern: kubectl rollout restart deployment/ kubectl_restart_match = re.search( r"kubectl\s+rollout\s+restart\s+deployment/([a-z0-9][\w.-]*)", action_lower ) if kubectl_restart_match: deploy_name = kubectl_restart_match.group(1) ns_match = re.search(r"-n\s+(\S+)", action_lower) namespace = ns_match.group(1) if ns_match else "default" return ParsedOperation(OperationType.RESTART_DEPLOYMENT, deploy_name, namespace) # Pattern: kubectl delete pod delete_pod_match = re.search( r"delete\s+pod[:\s]+([a-z0-9][\w.-]*)", action_lower ) if delete_pod_match: pod_name = delete_pod_match.group(1) ns_match = re.search(r"-n\s+(\S+)", action_lower) namespace = ns_match.group(1) if ns_match else "default" return ParsedOperation(OperationType.DELETE_POD, pod_name, namespace) # Pattern: 刪除 Pod (Chinese delete) chinese_delete_match = re.search(r"刪除\s*[Pp]od\s+([a-z0-9][\w.-]*)", action) if chinese_delete_match: pod_name = chinese_delete_match.group(1) return ParsedOperation(OperationType.DELETE_POD, pod_name, "default") # Pattern: restart deployment (English - with explicit "deployment") restart_deploy_match = re.search( r"restart\s+deployment[:\s]+([a-z0-9][\w.-]*)", action_lower ) if restart_deploy_match: deploy_name = restart_deploy_match.group(1) ns_match = re.search(r"-n\s+(\S+)", action_lower) namespace = ns_match.group(1) if ns_match else "default" return ParsedOperation(OperationType.RESTART_DEPLOYMENT, deploy_name, namespace) # Pattern: restart (English - without "deployment" keyword) restart_simple_match = re.search(r"restart\s+([a-z0-9][\w.-]*)", action_lower) if restart_simple_match: deploy_name = restart_simple_match.group(1) # Skip if captured word is "deployment" (handled above) if deploy_name != "deployment": ns_match = re.search(r"-n\s+(\S+)", action_lower) namespace = ns_match.group(1) if ns_match else "default" return ParsedOperation( OperationType.RESTART_DEPLOYMENT, deploy_name, namespace ) # Pattern: 重新啟動 deployment (Chinese with "deployment" keyword) chinese_restart_deploy_match = re.search( r"重新啟動\s+deployment\s+([a-z0-9][\w.-]*)", action, re.IGNORECASE ) if chinese_restart_deploy_match: deploy_name = chinese_restart_deploy_match.group(1) return ParsedOperation(OperationType.RESTART_DEPLOYMENT, deploy_name, "default") # Pattern: 重新啟動 服務 (Chinese) chinese_restart_match = re.search(r"重新啟動\s+([a-z0-9][\w.-]*)\s*服務", action) if chinese_restart_match: resource_name = chinese_restart_match.group(1) # StatefulSet Pod 格式: name-N (如 postgres-primary-0) if re.match(r".*-\d+$", resource_name): return ParsedOperation(OperationType.DELETE_POD, resource_name, "default") return ParsedOperation( OperationType.RESTART_DEPLOYMENT, resource_name, "default" ) # Pattern: scale deployment scale_match = re.search( r"scale\s+(?:deployment[:\s]+)?([a-z0-9][\w.-]*)", action_lower ) if scale_match: deploy_name = scale_match.group(1) ns_match = re.search(r"-n\s+(\S+)", action_lower) namespace = ns_match.group(1) if ns_match else "default" return ParsedOperation(OperationType.SCALE_DEPLOYMENT, deploy_name, namespace) # Pattern: 擴容 (Chinese scale) chinese_scale_match = re.search(r"擴容\s+([a-z0-9][\w.-]*)", action) if chinese_scale_match: deploy_name = chinese_scale_match.group(1) return ParsedOperation(OperationType.SCALE_DEPLOYMENT, deploy_name, "default") # Pattern: 擴展 副本數 (Chinese scale variant) chinese_scale2_match = re.search(r"擴展\s+([a-z0-9][\w.-]*)\s*副本", action) if chinese_scale2_match: deploy_name = chinese_scale2_match.group(1) # 移除常見的後綴如 -deployment deploy_name = re.sub(r"-deployment$", "", deploy_name) return ParsedOperation(OperationType.SCALE_DEPLOYMENT, deploy_name, "default") # Pattern: 重新啟動 (Chinese restart without 服務) chinese_restart2_match = re.search(r"重新啟動\s+([a-z0-9][\w.-]*)", action) if chinese_restart2_match: resource_name = chinese_restart2_match.group(1) # StatefulSet Pod 格式: name-N (如 postgres-primary-0) if re.match(r".*-\d+$", resource_name): return ParsedOperation(OperationType.DELETE_POD, resource_name, "default") # 移除常見的後綴 deploy_name = re.sub(r"-deployment$", "", resource_name) return ParsedOperation( OperationType.RESTART_DEPLOYMENT, deploy_name, "default" ) return ParsedOperation(None, None, "default")