""" Abstract Parser 抽象解析器基类 """ import json import yaml from abc import ABC, abstractmethod from typing import Dict, Any, Optional from app.module.operator.schema import OperatorDto, OperatorReleaseDto from app.module.operator.constants import CATEGORY_MAP, CATEGORY_OTHER_VENDOR_ID, CATEGORY_CUSTOMIZED_ID, \ CATEGORY_CLEANING_ID from app.module.operator.exceptions import FieldNotFoundError class AbstractParser(ABC): """算子文件解析器抽象基类""" @abstractmethod def parse_yaml_from_archive( self, archive_path: str, entry_path: str, file_name: Optional[str] = None, file_size: Optional[int] = None ) -> OperatorDto: """ 从压缩包内读取指定路径的 yaml 文件并解析为 OperatorDto Args: archive_path: 压缩包路径(zip 或 tar) entry_path: 压缩包内部的文件路径,例如 "config/app.yaml" Returns: 解析后的 OperatorDto """ pass @abstractmethod def extract_to(self, archive_path: str, target_dir: str) -> None: """ 将压缩包解压到目标目录(保持相对路径) Args: archive_path: 压缩包路径 target_dir: 目标目录 """ pass def parse_yaml( self, yaml_content: str, file_name: Optional[str] = None, file_size: Optional[int] = None ) -> OperatorDto: """解析 YAML 内容为 OperatorDto""" content: Dict[str, Any] = yaml.safe_load(yaml_content) operator = OperatorDto( id=self._to_string(content.get("raw_id")), name=self._to_string(content.get("name")), description=self._to_string(content.get("description")), version=self._to_string(content.get("version")), inputs=self._to_json(content.get("inputs")), outputs=self._to_json(content.get("outputs")), runtime=self._to_json(content.get("runtime")), settings=self._to_json(content.get("settings")), metrics=self._to_json(content.get("metrics")), file_name=file_name, file_size=file_size, ) # Handle changelog changelog = content.get("release") if isinstance(changelog, list): operator_release = OperatorReleaseDto( id=operator.id, version=operator.version, changelog=changelog ) else: operator_release = OperatorReleaseDto( id=operator.id, version=operator.version, changelog=[] ) operator.releases = [operator_release] # Build categories categories = [] types = content.get("types") if isinstance(types, list): for t in types: if self._to_lower(t) in CATEGORY_MAP: categories.append(CATEGORY_MAP[self._to_lower(t)]) if len(categories) == 0: categories.append(CATEGORY_CLEANING_ID) categories.extend([ CATEGORY_MAP.get(self._to_lower(content.get("language")), ""), CATEGORY_MAP.get(self._to_lower(content.get("modal")), ""), CATEGORY_MAP.get(self._to_lower(content.get("vendor")), CATEGORY_OTHER_VENDOR_ID), CATEGORY_CUSTOMIZED_ID, ]) operator.categories = categories return operator def _to_string(self, obj: Any) -> str: """转换为字符串""" if obj is None: raise FieldNotFoundError("field") return str(obj) def _to_lower(self, obj: Any) -> str: """转换为小写字符串""" if obj is None: raise FieldNotFoundError("field") return str(obj).lower() def _to_json(self, obj: Any) -> Optional[str]: """转换为 JSON 字符串""" if obj is None: return None try: return json.dumps(obj).strip('"').strip("'") except (TypeError, ValueError) as e: raise ValueError(f"Failed to serialize to JSON: {e}")