zkxw/backend/app/crud/sample_storage_cert_qc_acl.py

"""储存证质检字段 ACL CRUD。"""
from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Set

from sqlalchemy.orm import Session

from app.core.db_collation import eq_unicode, unicode_ci
from app.core.utils import generate_uuid
from app.crud.child_row_sync import sync_child_rows
from app.models.models import BizStorageCertQcAcl, BizStorageCertQcAclUser, SysUser
from app.services.sample_storage_cert_qc import (
    ACL_SCOPES,
    ALL_STORAGE_CERT_QC_FIELD_KEYS,
    dumps_field_keys,
    legal_field_keys_for_type,
    loads_field_keys,
)


def is_acl_strict(db: Session, tenant_no: str) -> bool:
    row = (
        db.query(BizStorageCertQcAcl.id)
        .filter(
            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
            eq_unicode(BizStorageCertQcAcl.configured_flag, "1"),
            BizStorageCertQcAcl.del_flag == 0,
        )
        .first()
    )
    return row is not None


def resolve_editable_field_keys(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    sample_type_key: Optional[str],
    has_legacy_write: bool,
) -> List[str]:
    legal = legal_field_keys_for_type(sample_type_key)
    if not is_acl_strict(db, tenant_no):
        return sorted(legal) if has_legacy_write else []
    uid = (user_id or "").strip()
    if not uid:
        return []
    acls = (
        db.query(BizStorageCertQcAcl)
        .filter(
            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
            eq_unicode(BizStorageCertQcAcl.configured_flag, "1"),
            BizStorageCertQcAcl.del_flag == 0,
        )
        .all()
    )
    if not acls:
        return []
    acl_ids = [row.id for row in acls]
    membership = (
        db.query(BizStorageCertQcAclUser.acl_id)
        .filter(
            eq_unicode(BizStorageCertQcAclUser.tenant_no, tenant_no),
            unicode_ci(BizStorageCertQcAclUser.acl_id).in_(acl_ids),
            eq_unicode(BizStorageCertQcAclUser.user_id, uid),
            BizStorageCertQcAclUser.del_flag == 0,
        )
        .all()
    )
    member_acl_ids = {row[0] for row in membership}
    keys: Set[str] = set()
    for acl in acls:
        if acl.id not in member_acl_ids:
            continue
        keys.update(loads_field_keys(acl.field_keys))
    return sorted(keys & legal)


def _validate_scope(scope: str) -> str:
    value = (scope or "").strip()
    if value not in ACL_SCOPES:
        raise ValueError("scope 必须为 qc 或 production")
    return value


def _validate_field_keys(field_keys: Any) -> List[str]:
    if not isinstance(field_keys, list):
        raise ValueError("fieldKeys 必须为数组")
    out: List[str] = []
    seen = set()
    for item in field_keys:
        key = str(item or "").strip()
        if not key or key in seen:
            continue
        if key not in ALL_STORAGE_CERT_QC_FIELD_KEYS:
            raise ValueError(f"非法字段: {key}")
        seen.add(key)
        out.append(key)
    return out


def _validate_user_ids(db: Session, tenant_no: str, user_ids: Any) -> List[str]:
    if not isinstance(user_ids, list):
        raise ValueError("userIds 必须为数组")
    out: List[str] = []
    seen = set()
    for item in user_ids:
        uid = str(item or "").strip()
        if not uid or uid in seen:
            continue
        seen.add(uid)
        out.append(uid)
    if not out:
        return []
    rows = (
        db.query(SysUser.id)
        .filter(
            eq_unicode(SysUser.tenant_no, tenant_no),
            unicode_ci(SysUser.id).in_(out),
            SysUser.del_flag == 0,
        )
        .all()
    )
    found = {row[0] for row in rows}
    missing = [uid for uid in out if uid not in found]
    if missing:
        raise ValueError(f"用户不存在或不属于当前租户: {','.join(missing[:5])}")
    return out


def get_acl_detail(db: Session, tenant_no: str, scope: str) -> Dict[str, Any]:
    scope_key = _validate_scope(scope)
    row = (
        db.query(BizStorageCertQcAcl)
        .filter(
            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
            eq_unicode(BizStorageCertQcAcl.scope, scope_key),
            BizStorageCertQcAcl.del_flag == 0,
        )
        .first()
    )
    if not row:
        return {
            "scope": scope_key,
            "fieldKeys": [],
            "userIds": [],
            "users": [],
            "configuredFlag": False,
            "aclStrict": is_acl_strict(db, tenant_no),
        }
    users = (
        db.query(BizStorageCertQcAclUser, SysUser)
        .outerjoin(
            SysUser,
            eq_unicode(SysUser.id, BizStorageCertQcAclUser.user_id) & (SysUser.del_flag == 0),
        )
        .filter(
            eq_unicode(BizStorageCertQcAclUser.tenant_no, tenant_no),
            eq_unicode(BizStorageCertQcAclUser.acl_id, row.id),
            BizStorageCertQcAclUser.del_flag == 0,
        )
        .order_by(BizStorageCertQcAclUser.order_num.asc(), BizStorageCertQcAclUser.create_time.asc())
        .all()
    )
    user_ids: List[str] = []
    user_rows: List[Dict[str, str]] = []
    for link, user in users:
        uid = link.user_id or ""
        if not uid:
            continue
        user_ids.append(uid)
        user_rows.append(
            {
                "id": uid,
                "nickName": (user.nick_name if user else "") or "",
                "userName": (user.user_no if user else "") or "",
            }
        )
    return {
        "scope": scope_key,
        "fieldKeys": loads_field_keys(row.field_keys),
        "userIds": user_ids,
        "users": user_rows,
        "configuredFlag": str(row.configured_flag or "0") == "1",
        "aclStrict": is_acl_strict(db, tenant_no),
    }


def upsert_acl(
    db: Session,
    *,
    tenant_no: str,
    user_id: str,
    scope: str,
    field_keys: Any,
    user_ids: Any,
) -> Dict[str, Any]:
    scope_key = _validate_scope(scope)
    keys = _validate_field_keys(field_keys)
    users = _validate_user_ids(db, tenant_no, user_ids)
    now = datetime.now()
    row = (
        db.query(BizStorageCertQcAcl)
        .filter(
            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
            eq_unicode(BizStorageCertQcAcl.scope, scope_key),
            BizStorageCertQcAcl.del_flag == 0,
        )
        .first()
    )
    if not row:
        row = BizStorageCertQcAcl(
            id=generate_uuid(),
            scope=scope_key,
            field_keys=dumps_field_keys(keys),
            configured_flag="1",
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)
        db.flush()
    else:
        row.field_keys = dumps_field_keys(keys)
        row.configured_flag = "1"
        row.update_by = user_id
        row.update_time = now

    def _build_row(new_id: str):
        return BizStorageCertQcAclUser(
            id=new_id,
            acl_id=row.id,
            user_id="",
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )

    def _apply_fields(link, item: Dict[str, Any], index: int, ts: datetime):
        link.user_id = str(item.get("userId") or "").strip()
        link.tenant_no = tenant_no
        link.order_num = index
        link.update_by = user_id
        link.update_time = ts

    sync_child_rows(
        db,
        BizStorageCertQcAclUser,
        parent_attr="acl_id",
        parent_id=row.id,
        items=[{"userId": uid} for uid in users],
        build_row=_build_row,
        apply_fields=_apply_fields,
        tenant_no=tenant_no,
        user_id=user_id,
        now=now,
        natural_key=lambda item: str(item.get("userId") or "").strip(),
        row_natural_key=lambda link: str(getattr(link, "user_id", "") or "").strip(),
        duplicate_message="人员重复",
        foreign_message="人员记录不属于当前 ACL",
    )
    db.commit()
    return get_acl_detail(db, tenant_no, scope_key)