iTi-Flask/iti/exchange/base.py

from __future__ import annotations

from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, Protocol, Sequence


class ExchangeOperation(str, Enum):
    IMPORT = "import"
    EXPORT = "export"


class ExchangeTemplateSourceKind(str, Enum):
    LOCAL = "local"
    REMOTE = "remote"
    MAPPING = "mapping"
    CUSTOM = "custom"


@dataclass(frozen=True)
class ExchangeScope:
    biz_domain: str
    biz_obj: str
    operation: ExchangeOperation | str

    @classmethod
    def from_mapping(
        cls,
        *,
        biz_domain: str,
        biz_obj: str,
        operation: ExchangeOperation | str,
    ) -> "ExchangeScope":
        return cls(
            biz_domain=biz_domain,
            biz_obj=biz_obj,
            operation=operation,
        )

    def key(self) -> str:
        return f"{_slug_token(self.biz_domain)}:{_slug_token(self.biz_obj)}:{_slug_token(_operation_value(self.operation))}"

    def code(self) -> str:
        return ".".join(
            [
                _slug_token(self.biz_domain),
                _slug_token(self.biz_obj),
                _slug_token(_operation_value(self.operation)),
            ]
        )


@dataclass(frozen=True)
class ExchangeVariable:
    key: str
    label: str
    header: str | None = None
    description: str | None = None
    required: bool = False
    example: str | None = None

    def workbook_header(self) -> str:
        return self.header or self.label or self.key


@dataclass(frozen=True)
class ExchangeTemplateLayout:
    title: str | None = None
    sheet_name: str | None = None
    title_row: int | None = 1
    header_row: int = 2
    data_start_row: int | None = None


@dataclass(frozen=True)
class ExchangeTemplatePlan:
    scope: ExchangeScope
    code: str | None = None
    name: str | None = None
    description: str | None = None
    template_id: str | None = None
    version_id: str | None = None
    version: str | None = None
    layout: ExchangeTemplateLayout = field(default_factory=ExchangeTemplateLayout)
    variables: tuple[ExchangeVariable, ...] = ()

    @classmethod
    def from_mapping(
        cls,
        *,
        biz_domain: str,
        biz_obj: str,
        operation: ExchangeOperation | str,
        code: str | None = None,
        name: str | None = None,
        description: str | None = None,
        template_id: str | None = None,
        version_id: str | None = None,
        version: str | None = None,
        layout: ExchangeTemplateLayout | dict[str, Any] | None = None,
        variables: Sequence[ExchangeVariable] | None = None,
    ) -> "ExchangeTemplatePlan":
        return cls(
            scope=ExchangeScope.from_mapping(
                biz_domain=biz_domain,
                biz_obj=biz_obj,
                operation=operation,
            ),
            code=code,
            name=name,
            description=description,
            template_id=template_id,
            version_id=version_id,
            version=version,
            layout=_coerce_layout(layout),
            variables=tuple(variables or ()),
        )

    def generated_code(self) -> str:
        return self.code or self.scope.code()

    def to_snapshot(
        self,
        *,
        published_at: str | None = None,
        file_key: str | None = None,
        checksum: str | None = None,
    ) -> "ExchangeTemplateSnapshot":
        return ExchangeTemplateSnapshot(
            scope=self.scope,
            code=self.code,
            name=self.name,
            description=self.description,
            template_id=self.template_id,
            version_id=self.version_id,
            version=self.version,
            layout=self.layout,
            variables=self.variables,
            published_at=published_at,
            file_key=file_key,
            checksum=checksum,
        )

    def as_payload(self) -> dict[str, Any]:
        return {
            "scope": _scope_payload(self.scope),
            "code": self.generated_code(),
            "name": self.name,
            "description": self.description,
            "template_id": self.template_id,
            "version_id": self.version_id,
            "version": self.version,
            "layout": asdict(self.layout),
            "variables": [asdict(item) for item in self.variables],
        }


@dataclass(frozen=True)
class ExchangeTemplateSnapshot(ExchangeTemplatePlan):
    published_at: str | None = None
    file_key: str | None = None
    checksum: str | None = None

    def as_payload(self) -> dict[str, Any]:
        payload = super().as_payload()
        payload.update(
            {
                "published_at": self.published_at,
                "file_key": self.file_key,
                "checksum": self.checksum,
            }
        )
        return payload


@dataclass(frozen=True)
class ExchangeBusinessSpec:
    scope: ExchangeScope
    name: str
    description: str | None = None
    layout: ExchangeTemplateLayout = field(default_factory=ExchangeTemplateLayout)
    variables: tuple[ExchangeVariable, ...] = ()
    code: str | None = None
    handler_name: str | None = None

    def generated_code(self) -> str:
        return self.code or self.scope.code()

    def to_plan(
        self,
        *,
        template_id: str | None = None,
        version_id: str | None = None,
        version: str | None = None,
    ) -> ExchangeTemplatePlan:
        return ExchangeTemplatePlan(
            scope=self.scope,
            code=self.generated_code(),
            name=self.name,
            description=self.description,
            template_id=template_id,
            version_id=version_id,
            version=version,
            layout=self.layout,
            variables=self.variables,
        )

    def as_payload(self) -> dict[str, Any]:
        payload = {
            "scope": _scope_payload(self.scope),
            "code": self.generated_code(),
            "name": self.name,
            "description": self.description,
            "layout": asdict(self.layout),
            "variables": [asdict(item) for item in self.variables],
        }
        if self.handler_name is not None:
            payload["handler_name"] = self.handler_name
        return payload


@dataclass(frozen=True)
class ExchangeTaskContext:
    task_id: str
    plan: ExchangeTemplatePlan
    snapshot: ExchangeTemplateSnapshot | None = None
    storage_key: str | None = None
    payload: dict[str, Any] = field(default_factory=dict)
    requested_by: str | None = None


@dataclass(frozen=True)
class ExchangeTaskResult:
    success_count: int = 0
    failed_count: int = 0
    message: str | None = None
    result_payload: dict[str, Any] = field(default_factory=dict)


class ExchangeTaskHandler(Protocol):
    def __call__(self, context: ExchangeTaskContext) -> ExchangeTaskResult:
        ...


class DataExchangeModule(Protocol):
    name: str

    def init_app(self, app) -> None:
        ...

    def register_routes(self, app) -> None:
        ...

    def register_permissions(self, app) -> None:
        ...

    def register_menu_seed(self, app) -> None:
        ...

    def register_tasks(self, app) -> None:
        ...


def _coerce_layout(value: ExchangeTemplateLayout | dict[str, Any] | None) -> ExchangeTemplateLayout:
    if value is None:
        return ExchangeTemplateLayout()
    if isinstance(value, ExchangeTemplateLayout):
        return value
    return ExchangeTemplateLayout(
        title=value.get("title"),
        sheet_name=value.get("sheet_name") or value.get("sheetName"),
        title_row=value.get("title_row", value.get("titleRow", 1)),
        header_row=value.get("header_row", value.get("headerRow", 2)),
        data_start_row=value.get("data_start_row") or value.get("dataStartRow"),
    )


def _operation_value(value: ExchangeOperation | str) -> str:
    return value.value if isinstance(value, ExchangeOperation) else str(value)


def _scope_payload(scope: ExchangeScope) -> dict[str, str]:
    return {
        "biz_domain": scope.biz_domain,
        "biz_obj": scope.biz_obj,
        "operation": _operation_value(scope.operation),
    }


def _slug_token(value: str) -> str:
    normalized = []
    previous_underscore = False
    for char in str(value).strip().lower():
        if char.isalnum():
            normalized.append(char)
            previous_underscore = False
            continue
        if char in {"_", "-"}:
            if not previous_underscore:
                normalized.append("_")
                previous_underscore = True
            continue
        if not previous_underscore:
            normalized.append("_")
            previous_underscore = True
    token = "".join(normalized).strip("_")
    return token or "item"