company-ai-platform/app/modules/legacy_mysql/services/mappers.py

from typing import Any


from app.core.config import get_settings
from app.modules.business.constants import SourceSystem, StatusValue
from app.modules.legacy_mysql.constants import (
    LEGACY_PROJECT_CODE_TEMPLATE,
    LEGACY_TASK_CODE_TEMPLATE,
    LEGACY_UNNAMED_PROJECT,
    LEGACY_UNNAMED_TASK,
    LegacyProjectField,
    LegacyTaskField,
)


class LegacyMapperMixin:
    @staticmethod
    def _value(
        row: dict[str, Any],
        field_map: dict[str, str],
        internal_name: str,
        fallback: Any = None,
    ) -> Any:
        source_name = field_map.get(internal_name, internal_name)
        if source_name in row:
            return row[source_name]
        return fallback

    def _project_payload(self, row: dict[str, Any], field_map: dict[str, str]) -> dict[str, Any]:
        settings = get_settings()
        external_id = self._value(
            row,
            field_map,
            LegacyProjectField.EXTERNAL_ID,
            row.get(LegacyProjectField.ID),
        )
        raw_code = self._value(row, field_map, LegacyProjectField.CODE, None)
        code = None
        if raw_code:
            code = str(raw_code)
        elif external_id is not None:
            code = LEGACY_PROJECT_CODE_TEMPLATE.format(
                prefix=settings.legacy_project_code_prefix,
                external_id=external_id,
            )
        return {
            LegacyProjectField.CODE: code,
            LegacyProjectField.EXTERNAL_ID: str(external_id) if external_id is not None else code,
            LegacyProjectField.SOURCE_SYSTEM: SourceSystem.LEGACY_MYSQL,
            LegacyProjectField.NAME: self._value(
                row,
                field_map,
                LegacyProjectField.NAME,
                LEGACY_UNNAMED_PROJECT,
            ),
            LegacyProjectField.OWNER: self._value(row, field_map, LegacyProjectField.OWNER, None),
            LegacyProjectField.STATUS: self._value(row, field_map, LegacyProjectField.STATUS, StatusValue.UNKNOWN),
            LegacyProjectField.PROGRESS_PERCENT: int(
                self._value(
                    row,
                    field_map,
                    LegacyProjectField.PROGRESS_PERCENT,
                    row.get(LegacyProjectField.PROGRESS) or 0,
                )
                or 0
            ),
            LegacyProjectField.START_DATE: self._value(row, field_map, LegacyProjectField.START_DATE, None),
            LegacyProjectField.DUE_DATE: self._value(row, field_map, LegacyProjectField.DUE_DATE, None),
            LegacyProjectField.BUDGET_AMOUNT: (
                self._value(row, field_map, LegacyProjectField.BUDGET_AMOUNT, row.get(LegacyProjectField.BUDGET) or 0) or 0
            ),
            LegacyProjectField.ACTUAL_AMOUNT: (
                self._value(row, field_map, LegacyProjectField.ACTUAL_AMOUNT, row.get(LegacyProjectField.ACTUAL_COST) or 0) or 0
            ),
            LegacyProjectField.DESCRIPTION: self._value(row, field_map, LegacyProjectField.DESCRIPTION, None),
        }

    def _task_payload(self, row: dict[str, Any], field_map: dict[str, str]) -> dict[str, Any]:
        settings = get_settings()
        external_id = self._value(
            row,
            field_map,
            LegacyTaskField.EXTERNAL_ID,
            row.get(LegacyTaskField.ID),
        )
        raw_code = self._value(row, field_map, LegacyTaskField.CODE, None)
        code = None
        if raw_code:
            code = str(raw_code)
        elif external_id is not None:
            code = LEGACY_TASK_CODE_TEMPLATE.format(
                prefix=settings.legacy_task_code_prefix,
                external_id=external_id,
            )
        return {
            LegacyTaskField.CODE: code,
            LegacyTaskField.EXTERNAL_ID: str(external_id) if external_id is not None else code,
            LegacyTaskField.SOURCE_SYSTEM: SourceSystem.LEGACY_MYSQL,
            LegacyTaskField.TITLE: self._value(
                row,
                field_map,
                LegacyTaskField.TITLE,
                LEGACY_UNNAMED_TASK,
            ),
            LegacyTaskField.PROJECT_CODE: self._value(
                row,
                field_map,
                LegacyTaskField.PROJECT_CODE,
                None,
            ),
            LegacyTaskField.OWNER: self._value(row, field_map, LegacyTaskField.OWNER, None),
            LegacyTaskField.STATUS: self._value(
                row,
                field_map,
                LegacyTaskField.STATUS,
                StatusValue.TODO,
            ),
            LegacyTaskField.PRIORITY: self._value(
                row,
                field_map,
                LegacyTaskField.PRIORITY,
                "P2",
            ),
            LegacyTaskField.DUE_DATE: self._value(row, field_map, LegacyTaskField.DUE_DATE, None),
            LegacyTaskField.COMPLETED_AT: self._value(
                row,
                field_map,
                LegacyTaskField.COMPLETED_AT,
                None,
            ),
            LegacyTaskField.BLOCKER: self._value(row, field_map, LegacyTaskField.BLOCKER, None),
            LegacyTaskField.DESCRIPTION: self._value(
                row,
                field_map,
                LegacyTaskField.DESCRIPTION,
                None,
            ),
        }