agent-platform/services/skill-service/app/schemas/skill.py

from datetime import datetime
from typing import TYPE_CHECKING, Generic, TypeVar

from core_domain import (
    SkillDefinitionContract,
    SkillInstallationContract,
    SkillInstallStatus,
    SkillRunContract,
    SkillStatus,
)
from core_shared import JSONValue
from pydantic import BaseModel, Field

if TYPE_CHECKING:
    from app.db.models import SkillDefinition, SkillInstallation, SkillRun

T = TypeVar("T")


class ApiErrorResponse(BaseModel):
    errorType: str
    message: str
    details: dict[str, JSONValue] = Field(default_factory=dict)


class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    data: T | None = None
    error: ApiErrorResponse | None = None
    requestId: str
    serverTime: datetime


class PageRequest(BaseModel):
    page: int = Field(default=1, ge=1)
    pageSize: int = Field(default=20, ge=1, le=200)
    keyword: str | None = None

    @property
    def offset(self) -> int:
        return (self.page - 1) * self.pageSize


class PageResult(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    pageSize: int
    hasMore: bool

    @classmethod
    def from_items(
        cls,
        *,
        items: list[T],
        total: int,
        page: int,
        page_size: int) -> "PageResult[T]":
        return cls(
            items=items,
            total=total,
            page=page,
            pageSize=page_size,
            hasMore=page * page_size < total)


class SkillCreateRequest(BaseModel):
    code: str
    name: str
    skill_type: str = "template"
    description: str | None = None
    runtime_type: str = "template"
    entrypoint: str | None = None
    parameter_schema_json: dict[str, JSONValue] = Field(default_factory=dict)
    output_schema_json: dict[str, JSONValue] = Field(default_factory=dict)
    implementation_json: dict[str, JSONValue] = Field(default_factory=dict)
    owner_user_id: str | None = None
    metadata_json: dict[str, JSONValue] = Field(default_factory=dict)


class SkillStatusUpdateRequest(BaseModel):
    status: SkillStatus


class SkillResponse(SkillDefinitionContract):
    @classmethod
    def from_entity(cls, entity: "SkillDefinition") -> "SkillResponse":
        return cls.model_validate(entity, from_attributes=True)


class SkillInstallRequest(BaseModel):
    skill_id: str
    install_scope: str = "global"
    scope_id: str
    config_json: dict[str, JSONValue] = Field(default_factory=dict)
    installed_by: str | None = None


class SkillInstallationStatusUpdateRequest(BaseModel):
    status: SkillInstallStatus


class SkillInstallationResponse(SkillInstallationContract):
    @classmethod
    def from_entity(cls, entity: "SkillInstallation") -> "SkillInstallationResponse":
        return cls.model_validate(entity, from_attributes=True)


class SkillRunCreateRequest(BaseModel):
    skill_id: str
    installation_id: str | None = None
    input_json: dict[str, JSONValue] = Field(default_factory=dict)


class SkillRunExecuteRequest(BaseModel):
    skillRunId: str | None = None
    worker_key: str | None = None


class SkillRunResponse(SkillRunContract):
    @classmethod
    def from_entity(cls, entity: "SkillRun") -> "SkillRunResponse":
        return cls.model_validate(entity, from_attributes=True)


class SkillRunDto(BaseModel):
    id: str
    skillId: str
    installationId: str | None = None
    status: str
    input: dict[str, JSONValue] = Field(default_factory=dict)
    output: dict[str, JSONValue] | None = None
    outputText: str | None = None
    workerKey: str | None = None
    startedTime: datetime | None = None
    finishedTime: datetime | None = None
    errorCode: str | None = None
    errorMessage: str | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "SkillRun") -> "SkillRunDto":
        return cls(
            id=entity.id,
            skillId=entity.skill_id,
            installationId=entity.installation_id,
            status=entity.status,
            input=entity.input_json or {},
            output=entity.output_json,
            outputText=entity.output_text,
            workerKey=entity.worker_key,
            startedTime=entity.started_time,
            finishedTime=entity.finished_time,
            errorCode=entity.error_code,
            errorMessage=entity.error_message,
            createdTime=entity.created_time)


class SkillRunCreateRequestDto(BaseModel):
    skillId: str
    installationId: str | None = None
    input: dict[str, JSONValue] = Field(default_factory=dict)


class SkillRunExecuteRequestDto(BaseModel):
    skillRunId: str
    workerKey: str | None = None


class SkillDefinitionDto(BaseModel):
    id: str
    name: str
    skillType: str
    description: str | None = None
    status: SkillStatus
    ownerUserId: str | None = None
    metadata: dict[str, JSONValue] = Field(default_factory=dict)
    category: str = "service"
    instruction: str = ""
    toolIds: list[str] = Field(default_factory=list)
    runtimeType: str = "template"
    entrypoint: str | None = None
    parameterSchema: dict[str, JSONValue] = Field(default_factory=dict)
    outputSchema: dict[str, JSONValue] = Field(default_factory=dict)
    implementation: dict[str, JSONValue] = Field(default_factory=dict)
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "SkillDefinition") -> "SkillDefinitionDto":
        metadata = dict(entity.metadata_json or {})
        tool_ids = metadata.get("toolIds")
        return cls(
            id=entity.id,
            name=entity.name,
            skillType=entity.skill_type,
            description=entity.description,
            status=entity.status,
            ownerUserId=entity.owner_user_id,
            metadata=metadata,
            category=str(metadata.get("category") or "service"),
            instruction=str(metadata.get("instruction") or ""),
            toolIds=[str(item) for item in tool_ids] if isinstance(tool_ids, list) else [],
            runtimeType=entity.runtime_type,
            entrypoint=entity.entrypoint,
            parameterSchema=entity.parameter_schema_json,
            outputSchema=entity.output_schema_json,
            implementation=entity.implementation_json,
            createdTime=entity.created_time)


class SkillListRequestDto(PageRequest):
    status: SkillStatus | None = None
    skillType: str | None = None
    category: str | None = None


class SkillCreateRequestDto(BaseModel):
    name: str
    skillType: str = "template"
    description: str | None = None
    category: str = "service"
    instruction: str = ""
    toolIds: list[str] = Field(default_factory=list)
    runtimeType: str = "template"
    entrypoint: str | None = None
    parameterSchema: dict[str, JSONValue] = Field(default_factory=dict)
    outputSchema: dict[str, JSONValue] = Field(default_factory=dict)
    implementation: dict[str, JSONValue] = Field(default_factory=dict)
    ownerUserId: str | None = None
    metadata: dict[str, JSONValue] = Field(default_factory=dict)


class SkillDetailRequestDto(BaseModel):
    skillId: str


class SkillUpdateRequestDto(BaseModel):
    skillId: str
    name: str | None = None
    skillType: str | None = None
    description: str | None = None
    category: str | None = None
    instruction: str | None = None
    toolIds: list[str] | None = None
    runtimeType: str | None = None
    entrypoint: str | None = None
    parameterSchema: dict[str, JSONValue] | None = None
    outputSchema: dict[str, JSONValue] | None = None
    implementation: dict[str, JSONValue] | None = None
    ownerUserId: str | None = None
    metadata: dict[str, JSONValue] | None = None


class SkillStatusRequestDto(BaseModel):
    skillId: str
    status: SkillStatus


class SkillDeleteRequestDto(BaseModel):
    skillId: str


class SkillInstallationDto(BaseModel):
    id: str
    skillId: str
    installScope: str
    scopeId: str | None = None
    status: SkillInstallStatus
    config: dict[str, JSONValue] = Field(default_factory=dict)
    installedBy: str | None = None
    installedTime: datetime | None = None
    createdTime: datetime

    @classmethod
    def from_entity(cls, entity: "SkillInstallation") -> "SkillInstallationDto":
        return cls(
            id=entity.id,
            skillId=entity.skill_id,
            installScope=entity.install_scope,
            scopeId=entity.scope_id,
            status=entity.status,
            config=entity.config_json or {},
            installedBy=entity.installed_by,
            installedTime=entity.installed_time,
            createdTime=entity.created_time)


class SkillInstallationListRequestDto(PageRequest):
    installScope: str | None = None
    scopeId: str | None = None
    status: SkillInstallStatus | None = None


class SkillInstallRequestDto(BaseModel):
    skillId: str
    installScope: str = "global"
    scopeId: str = "global"
    config: dict[str, JSONValue] = Field(default_factory=dict)
    installedBy: str | None = None


class SkillInstallationStatusRequestDto(BaseModel):
    installationId: str
    status: SkillInstallStatus


class DeleteData(BaseModel):
    deleted: bool
    skillId: str | None = None
    installationId: str | None = None