zkxw/backend/app/api/lab_sample_processing.py

"""实验室样本处理数据 API"""
from __future__ import annotations

import csv
import io
from typing import Any, Dict, Optional

from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
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_strict_any_menu_code
from app.core.response import error_response, success_response
from app.models.models import SysUser
from app.services import lab_sample_processing as svc
from app.services.lab_sample_processing_fields import (
    DATASETS,
    DATASET_LABELS,
    PERM_LIST,
    PERM_MANAGER,
    PERM_MARKET_EDIT,
    PERM_PRODUCTION_EDIT,
    PERM_QC_EDIT,
    schema_payload,
)

router = APIRouter(prefix="/web/admin/lab-sample-processing", tags=["实验室样本处理数据"])

# 列表/导出：有列表、主管或任一部门编辑权即可查看全部字段
READ_MENUS = (
    PERM_LIST,
    PERM_MANAGER,
    PERM_MARKET_EDIT,
    PERM_PRODUCTION_EDIT,
    PERM_QC_EDIT,
)
# 更新：至少要有主管或某一部门编辑权（字段级 ACL 在服务内再收紧）
WRITE_MENUS = (
    PERM_MANAGER,
    PERM_MARKET_EDIT,
    PERM_PRODUCTION_EDIT,
    PERM_QC_EDIT,
)


def _tenant_no(db: Session, user: dict) -> str:
    user_id = user.get("user_id")
    if not user_id:
        raise PermissionError("用户身份信息缺失")
    row = db.query(SysUser).filter(SysUser.id == user_id, SysUser.del_flag == 0).first()
    if not row or not row.tenant_no:
        raise PermissionError("用户租户信息缺失")
    return row.tenant_no


def _ensure_dataset(dataset: str) -> Optional[str]:
    ds = (dataset or "").strip().lower()
    return ds if ds in DATASETS else None


class UpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    payload: Dict[str, Any] = Field(default_factory=dict)


@router.get("/meta/{dataset}")
async def get_meta(
    dataset: str,
    user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
    db: Session = Depends(get_db),
):
    ds = _ensure_dataset(dataset)
    if not ds:
        return error_response(400, "未知数据集")
    acl = svc.user_acl(db, user)
    return success_response(
        {
            "dataset": ds,
            "label": DATASET_LABELS.get(ds),
            "fields": schema_payload(ds),
            "acl": acl,
        }
    )


@router.get("/{dataset}/list")
async def list_dataset(
    dataset: str,
    keyword: Optional[str] = Query(None),
    dateFrom: Optional[str] = Query(None),
    dateTo: Optional[str] = Query(None),
    page: int = Query(1, ge=1),
    pageSize: int = Query(20, ge=1, le=200),
    user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
    db: Session = Depends(get_db),
):
    ds = _ensure_dataset(dataset)
    if not ds:
        return error_response(400, "未知数据集")
    tenant_no = _tenant_no(db, user)
    rows, total = svc.list_rows(
        db,
        tenant_no=tenant_no,
        dataset=ds,
        keyword=keyword,
        date_from=dateFrom,
        date_to=dateTo,
        page=page,
        page_size=pageSize,
    )
    acl = svc.user_acl(db, user)
    return success_response(
        {
            "list": rows,
            "total": total,
            "page": page,
            "pageSize": pageSize,
            "fields": schema_payload(ds),
            "acl": acl,
        }
    )


@router.put("/{dataset}/{sample_code}")
async def update_dataset_row(
    dataset: str,
    sample_code: str,
    body: UpdateBody,
    user: dict = Depends(require_strict_any_menu_code(*WRITE_MENUS)),
    db: Session = Depends(get_db),
):
    ds = _ensure_dataset(dataset)
    if not ds:
        return error_response(400, "未知数据集")
    tenant_no = _tenant_no(db, user)
    acl = svc.user_acl(db, user)
    editable = set(acl.get("editableDepts") or [])
    if not editable:
        return error_response(403, "无编辑权限")
    try:
        row = svc.update_row(
            db,
            tenant_no=tenant_no,
            dataset=ds,
            sample_code=sample_code,
            payload=body.payload or {},
            editable_depts=editable,
            operator_id=str(user.get("user_id") or ""),
        )
    except PermissionError as exc:
        return error_response(403, str(exc))
    except LookupError as exc:
        return error_response(404, str(exc))
    except ValueError as exc:
        return error_response(400, str(exc))
    return success_response(row)


@router.get("/{dataset}/export")
async def export_dataset(
    dataset: str,
    keyword: Optional[str] = Query(None),
    dateFrom: Optional[str] = Query(None),
    dateTo: Optional[str] = Query(None),
    user: dict = Depends(require_strict_any_menu_code(*READ_MENUS)),
    db: Session = Depends(get_db),
):
    ds = _ensure_dataset(dataset)
    if not ds:
        return error_response(400, "未知数据集")
    tenant_no = _tenant_no(db, user)
    headers, body = svc.export_rows(
        db,
        tenant_no=tenant_no,
        dataset=ds,
        keyword=keyword,
        date_from=dateFrom,
        date_to=dateTo,
    )
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(headers)
    for row in body:
        writer.writerow(["" if v is None else v for v in row])
    raw = ("\ufeff" + buf.getvalue()).encode("utf-8")
    filename = f"lab-sample-{ds}.csv"
    return StreamingResponse(
        io.BytesIO(raw),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )