opsxsql/backend/app/api/v1/endpoints/instance.py

import json
from typing import Optional, List

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database import get_db
from app.core.security import get_current_user, require_role
from app.models.user import User
from app.models.instance import Instance
from app.schemas.instance import InstanceCreate, InstanceUpdate, InstanceResponse, ConnectionTestResult
from app.services.instance_service import InstanceService
from app.core.crypto import encrypt_password, decrypt_password
from app.services.audit_log_service import AuditLogService

router = APIRouter(prefix="/instances", tags=["数据源管理"])


def _serialize_json_fields(instance: Instance) -> dict:
    """将 Instance 的 JSON text 字段反序列化为 dict/list，返回 Response dict"""
    d = {
        "id": instance.id,
        "name": instance.name,
        "db_type": instance.db_type,
        "host": instance.host,
        "port": instance.port,
        "username": instance.username,
        "database": instance.database,
        "env": instance.env,
        "description": instance.description,
        "status": instance.status,
        "version": instance.version,
        "access_mode": instance.access_mode or "readwrite",
        "workflow": json.loads(instance.workflow) if instance.workflow else None,
        "excluded_databases": json.loads(instance.excluded_databases) if instance.excluded_databases else None,
        "mask_fields": json.loads(instance.mask_fields) if instance.mask_fields else None,
        "created_at": instance.created_at,
    }
    return d


@router.get("", response_model=List[InstanceResponse])
async def list_instances(
    db_type: Optional[str] = None,
    env: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    stmt = select(Instance).order_by(Instance.id)
    if db_type:
        stmt = stmt.where(Instance.db_type == db_type)
    if env:
        stmt = stmt.where(Instance.env == env)
    result = await db.execute(stmt)
    instances = result.scalars().all()
    return [_serialize_json_fields(i) for i in instances]


@router.post("", response_model=InstanceResponse, status_code=201)
async def create_instance(
    data: InstanceCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    instance = Instance(
        name=data.name,
        db_type=data.db_type,
        host=data.host,
        port=data.port,
        username=data.username,
        password=encrypt_password(data.password),
        database=data.database,
        env=data.env,
        description=data.description,
        access_mode=data.access_mode,
        workflow=json.dumps([s.model_dump() for s in data.workflow], ensure_ascii=False) if data.workflow else None,
        excluded_databases=json.dumps(data.excluded_databases, ensure_ascii=False) if data.excluded_databases else None,
        mask_fields=json.dumps([m.model_dump() for m in data.mask_fields], ensure_ascii=False) if data.mask_fields else None,
    )
    db.add(instance)
    await db.flush()
    await AuditLogService.log(
        db, user_id=current_user.id, action="CREATE_INSTANCE",
        target_type="instance", target_id=instance.id,
        detail=f"创建数据源: {data.name} ({data.db_type}@{data.host}:{data.port})",
    )
    await db.commit()
    await db.refresh(instance)
    return _serialize_json_fields(instance)


@router.post("/test-connection", response_model=ConnectionTestResult)
async def test_new_connection(
    data: InstanceCreate,
    current_user: User = Depends(require_role("admin", "dba")),
):
    """测试连接参数（不保存实例）"""
    fake = Instance(
        name=data.name, db_type=data.db_type, host=data.host, port=data.port,
        username=data.username, password=data.password, database=data.database,
        env=data.env, access_mode=data.access_mode,
    )
    return await InstanceService.test_connection(fake)


@router.get("/{instance_id}", response_model=InstanceResponse)
async def get_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")
    return _serialize_json_fields(instance)


@router.put("/{instance_id}", response_model=InstanceResponse)
async def update_instance(
    instance_id: int,
    data: InstanceUpdate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")

    update_data = data.model_dump(exclude_unset=True)
    if "password" in update_data:
        update_data["password"] = encrypt_password(update_data["password"])
    # JSON fields
    if "workflow" in update_data:
        update_data["workflow"] = json.dumps([s if isinstance(s, dict) else s for s in update_data["workflow"]], ensure_ascii=False) if update_data["workflow"] else None
    if "excluded_databases" in update_data:
        update_data["excluded_databases"] = json.dumps(update_data["excluded_databases"], ensure_ascii=False) if update_data["excluded_databases"] else None
    if "mask_fields" in update_data:
        update_data["mask_fields"] = json.dumps([m if isinstance(m, dict) else m for m in update_data["mask_fields"]], ensure_ascii=False) if update_data["mask_fields"] else None

    for key, value in update_data.items():
        setattr(instance, key, value)
    await db.commit()
    await db.refresh(instance)
    return _serialize_json_fields(instance)


@router.delete("/{instance_id}")
async def delete_instance(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")

    await AuditLogService.log(
        db, user_id=current_user.id, action="DELETE_INSTANCE",
        target_type="instance", target_id=instance.id,
        detail=f"删除数据源: {instance.name}",
    )
    await db.delete(instance)
    await db.commit()
    return {"message": "数据源已删除"}


@router.post("/{instance_id}/test", response_model=ConnectionTestResult)
async def test_connection(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin", "dba")),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")

    instance.password = decrypt_password(instance.password)
    return await InstanceService.test_connection(instance)


@router.get("/{instance_id}/databases")
async def list_databases(
    instance_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")

    return await InstanceService.list_databases(instance)


@router.get("/{instance_id}/databases/{database}/tables")
async def list_tables(
    instance_id: int,
    database: str,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    result = await db.execute(select(Instance).where(Instance.id == instance_id))
    instance = result.scalar_one_or_none()
    if not instance:
        raise HTTPException(status_code=404, detail="数据源不存在")

    return await InstanceService.list_tables(instance, database)