company-ai-platform/app/modules/legacy_mysql/service.py

from datetime import date, datetime
from decimal import Decimal
from typing import Any

from fastapi import HTTPException, status
from sqlalchemy import inspect, text
from sqlalchemy.engine import Engine, RowMapping
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.core.constants import ActorValue
from app.core.config import get_settings
from app.core.database import legacy_engine
from app.core.pagination import bounded_limit
from app.core.time import utc_now
from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource, AuditStatus
from app.modules.audit.schemas import AuditLogCreate
from app.modules.audit.service import AuditService
from app.modules.business.constants import BusinessDomain, SourceSystem, StatusValue
from app.modules.business.models import LegacySyncRun, Project
from app.modules.business.service import serialize_model
from app.modules.legacy_mysql.constants import LegacyQueryError, LegacyQueryName

FORBIDDEN_SQL_TOKENS = {
    "insert",
    "update",
    "delete",
    "drop",
    "alter",
    "truncate",
    "create",
    "replace",
    "grant",
    "revoke",
}


def _jsonable(value: Any) -> Any:
    """Convert database scalar values into JSON-friendly values."""

    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        return float(value)
    return value


def _row_to_dict(row: RowMapping) -> dict[str, Any]:
    """Convert a SQLAlchemy row mapping to a serializable dictionary."""

    return {key: _jsonable(value) for key, value in row.items()}


def _normalize_sql(sql: str) -> str:
    return " ".join(sql.strip().rstrip(";").split()).lower()


def _query_name_text(query_name: str | LegacyQueryName | None) -> str:
    if query_name is None:
        return LegacyQueryName.PROJECTS.value
    if isinstance(query_name, LegacyQueryName):
        return query_name.value
    return str(query_name)


class LegacyMySQLService:
    """Read legacy MySQL data and sync projects into the internal ledger."""

    def __init__(self, db: Session | None):
        self.db = db

    @staticmethod
    def _ensure_engine() -> Engine:
        if legacy_engine is None:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail="LEGACY_DATABASE_URL is not configured",
            )
        return legacy_engine

    @staticmethod
    def _ensure_readonly(sql: str) -> None:
        stripped = sql.strip().lower()
        if not stripped.startswith("select"):
            raise HTTPException(status_code=400, detail="Only SELECT statements are allowed")
        tokens = {token.strip(" ;,\n\t") for token in stripped.replace("(", " ").split()}
        if tokens & FORBIDDEN_SQL_TOKENS:
            raise HTTPException(status_code=400, detail="Forbidden SQL token in readonly query")

    @staticmethod
    def _allowed_queries() -> dict[str, str]:
        settings = get_settings()
        queries = {
            _query_name_text(name): sql
            for name, sql in settings.legacy_allowed_queries.items()
        }
        if settings.legacy_project_query:
            queries.setdefault(LegacyQueryName.PROJECTS.value, settings.legacy_project_query)
        return queries

    def health(self) -> dict[str, str]:
        engine = self._ensure_engine()
        try:
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))
        except SQLAlchemyError as exc:
            raise HTTPException(status_code=503, detail=f"MySQL connection failed: {exc}") from exc
        return {"status": "ok"}

    def list_tables(self) -> list[str]:
        engine = self._ensure_engine()
        return sorted(inspect(engine).get_table_names())

    def describe_table(self, table_name: str) -> list[dict[str, Any]]:
        engine = self._ensure_engine()
        inspector = inspect(engine)
        if table_name not in inspector.get_table_names():
            raise HTTPException(status_code=404, detail="Table not found")
        columns = []
        for column in inspector.get_columns(table_name):
            columns.append(
                {
                    "name": column["name"],
                    "type": str(column["type"]),
                    "nullable": column.get("nullable", True),
                    "default": (
                        str(column.get("default"))
                        if column.get("default") is not None
                        else None
                    ),
                }
            )
        return columns

    def execute_readonly(
        self,
        sql: str,
        params: dict[str, Any] | None = None,
        limit: int = 100,
    ) -> dict[str, Any]:
        """Execute a readonly SQL statement only when it matches the allowlist."""

        normalized_sql = _normalize_sql(sql)
        for allowed_sql in self._allowed_queries().values():
            if _normalize_sql(allowed_sql) == normalized_sql:
                return self._execute_readonly_sql(allowed_sql, params, limit)
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=LegacyQueryError.QUERY_NOT_ALLOWED,
        )

    def execute_allowed_query(
        self,
        query_name: str | None,
        params: dict[str, Any] | None = None,
        limit: int = 100,
    ) -> dict[str, Any]:
        queries = self._allowed_queries()
        normalized_name = _query_name_text(query_name)
        sql = queries.get(normalized_name)
        if not sql:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=LegacyQueryError.QUERY_NOT_ALLOWED,
            )
        return self._execute_readonly_sql(sql, params, limit)

    def _execute_readonly_sql(
        self,
        sql: str,
        params: dict[str, Any] | None = None,
        limit: int = 100,
    ) -> dict[str, Any]:
        self._ensure_readonly(sql)
        engine = self._ensure_engine()
        params = dict(params or {})
        try:
            params["limit"] = bounded_limit(params.get("limit", limit))
        except (TypeError, ValueError) as exc:
            raise HTTPException(
                status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
                detail="Invalid readonly query limit",
            ) from exc
        limited_sql = sql
        if " limit " not in sql.lower():
            limited_sql = f"{sql.rstrip(';')} LIMIT :limit"
        with engine.connect() as conn:
            result = conn.execute(text(limited_sql), params)
            rows = [_row_to_dict(row) for row in result.mappings().all()]
        columns = list(rows[0].keys()) if rows else []
        return {"columns": columns, "rows": rows, "row_count": len(rows)}

    def fetch_default_projects(self, limit: int = 100) -> dict[str, Any]:
        settings = get_settings()
        if not settings.legacy_project_query:
            raise HTTPException(
                status_code=400,
                detail=LegacyQueryError.PROJECT_QUERY_NOT_CONFIGURED,
            )
        return self.execute_allowed_query(LegacyQueryName.PROJECTS, {"limit": limit}, limit=limit)

    @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, "external_id", row.get("id"))
        raw_code = self._value(row, field_map, "code", None)
        code = None
        if raw_code:
            code = str(raw_code)
        elif external_id is not None:
            code = f"{settings.legacy_project_code_prefix}-{external_id}"
        return {
            "code": code,
            "external_id": str(external_id) if external_id is not None else code,
            "source_system": SourceSystem.LEGACY_MYSQL,
            "name": self._value(row, field_map, "name", "未命名项目"),
            "owner": self._value(row, field_map, "owner", None),
            "status": self._value(row, field_map, "status", StatusValue.UNKNOWN),
            "progress_percent": int(
                self._value(
                    row,
                    field_map,
                    "progress_percent",
                    row.get("progress") or 0,
                )
                or 0
            ),
            "start_date": self._value(row, field_map, "start_date", None),
            "due_date": self._value(row, field_map, "due_date", None),
            "budget_amount": (
                self._value(row, field_map, "budget_amount", row.get("budget") or 0) or 0
            ),
            "actual_amount": (
                self._value(row, field_map, "actual_amount", row.get("actual_cost") or 0) or 0
            ),
            "description": self._value(row, field_map, "description", None),
        }

    def sync_projects(
        self,
        source_query: str | None = None,
        source_query_name: str | None = None,
        field_map: dict[str, str] | None = None,
        limit: int = 100,
        dry_run: bool = True,
        actor: str = ActorValue.API,
    ) -> dict[str, Any]:
        if self.db is None:
            raise HTTPException(
                status_code=503,
                detail="Application database session is not available",
            )

        query_name = source_query_name or LegacyQueryName.PROJECTS
        if source_query:
            rows = self.execute_readonly(source_query, {"limit": limit}, limit=limit)["rows"]
            query_ref = "allowlisted_inline_sql"
        else:
            rows = self.execute_allowed_query(query_name, {"limit": limit}, limit=limit)["rows"]
            query_ref = _query_name_text(query_name)
        field_map = field_map or {}
        created = 0
        updated = 0
        skipped = 0
        items: list[dict[str, Any]] = []

        for row in rows:
            payload = self._project_payload(row, field_map)
            if not payload["external_id"] and not payload["code"]:
                skipped += 1
                items.append(
                    {
                        "action": "skipped",
                        "reason": "missing external_id/code",
                        "source": row,
                    }
                )
                continue

            stmt = select(Project).where(
                Project.source_system == SourceSystem.LEGACY_MYSQL,
                Project.external_id == payload["external_id"],
            )
            record = self.db.execute(stmt).scalar_one_or_none()
            if record is None:
                record = self.db.execute(
                    select(Project).where(Project.code == payload["code"])
                ).scalar_one_or_none()

            if record is None:
                created += 1
                action = "create"
                result = payload
                if not dry_run:
                    record = Project(**payload)
                    self.db.add(record)
                    self.db.flush()
                    result = serialize_model(record)
            else:
                updated += 1
                action = "update"
                if not dry_run:
                    for key, value in payload.items():
                        setattr(record, key, value)
                    self.db.flush()
                    result = serialize_model(record)
                else:
                    result = payload
            items.append({"action": action, "project": result, "source": row})

        if not dry_run:
            self.db.commit()

        result = {
            "dry_run": dry_run,
            "created": created,
            "updated": updated,
            "skipped": skipped,
            "items": items,
        }
        sync_run = LegacySyncRun(
            code=f"SYNC-PROJECTS-{utc_now():%Y%m%d%H%M%S%f}",
            domain=BusinessDomain.PROJECTS,
            source_table="LEGACY_PROJECT_QUERY",
            status=StatusValue.DRY_RUN if dry_run else AuditStatus.SUCCESS,
            finished_at=utc_now(),
            created_count=created,
            updated_count=updated,
            skipped_count=skipped,
            note="Project sync from readonly legacy MySQL",
        )
        self.db.add(sync_run)
        self.db.commit()
        self.db.refresh(sync_run)
        result["sync_run_code"] = sync_run.code

        AuditService(self.db).log(
            AuditLogCreate(
                actor=actor,
                source=AuditSource.LEGACY_MYSQL,
                action=AuditAction.LEGACY_SYNC_PROJECTS,
                target_type=BusinessDomain.PROJECTS,
                risk_level=AuditRiskLevel.MEDIUM,
                request_payload={
                    "source_query": query_ref,
                    "field_map": field_map,
                    "limit": limit,
                    "dry_run": dry_run,
                },
                response_payload={
                    key: result[key] for key in ["dry_run", "created", "updated", "skipped"]
                },
            )
        )
        return result