abstract_parser.py

"""
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}")