zkxw/backend/app/api/sample_code_correction.py

"""样本编码全链路纠错 API"""
from typing import Optional

from fastapi import APIRouter, Depends
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.core.menu_guard import require_admin_or_super_admin
from app.core.response import error_response, success_response
from app.models.models import SysUser
from app.services import sample_code_correction as svc
from app.services.sample_code_correction import SampleCodeCorrectionError

router = APIRouter(prefix="/web/admin", tags=["样本编码纠错"])

_MENU = require_admin_or_super_admin


def _get_tenant_no(db: Session, user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失，无法确定租户")
    sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
    if not sys_user or not sys_user.tenant_no:
        raise PermissionError("用户租户信息缺失")
    return sys_user.tenant_no


def _get_user_id(user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失")
    return str(user_id)


class SampleCodeCorrectionBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    reception_source: str = Field(..., alias="receptionSource")
    old_sample_code: str = Field(..., alias="oldSampleCode")
    new_sample_code: str = Field(..., alias="newSampleCode")
    new_sample_code_id: Optional[str] = Field(None, alias="newSampleCodeId")
    reason: Optional[str] = None


@router.post("/sample-code-corrections/preview")
async def preview_sample_code_correction(
    body: SampleCodeCorrectionBody,
    db: Session = Depends(get_db),
    user: dict = Depends(_MENU),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        data = svc.preview_correction(
            db,
            tenant_no=tenant_no,
            reception_source=body.reception_source,
            old_sample_code=body.old_sample_code,
            new_sample_code=body.new_sample_code,
            new_sample_code_id=body.new_sample_code_id,
        )
        return success_response(data)
    except PermissionError as e:
        return error_response(str(e))
    except Exception as e:
        return error_response(f"预览失败: {e}")


@router.post("/sample-code-corrections/apply")
async def apply_sample_code_correction(
    body: SampleCodeCorrectionBody,
    db: Session = Depends(get_db),
    user: dict = Depends(_MENU),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        user_id = _get_user_id(user)
        data = svc.apply_correction(
            db,
            tenant_no=tenant_no,
            reception_source=body.reception_source,
            old_sample_code=body.old_sample_code,
            new_sample_code=body.new_sample_code,
            new_sample_code_id=body.new_sample_code_id,
            reason=body.reason,
            user_id=user_id,
        )
        return success_response(data, message="样本编码已更正")
    except SampleCodeCorrectionError as e:
        db.rollback()
        return error_response(str(e))
    except PermissionError as e:
        db.rollback()
        return error_response(str(e))
    except Exception as e:
        db.rollback()
        return error_response(f"更正失败: {e}")