zkxw/backend/app/crud/env_detect_record.py

"""环境检测登记 CRUD"""
from __future__ import annotations

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

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.utils import format_date, generate_uuid
from app.crud import env_detect_item as item_svc
from app.models.models import BizEnvDetectRecord, SysUser


def _split_csv(value: Optional[str]) -> List[str]:
    return [x.strip() for x in (value or "").split(",") if x.strip()]


def _join_csv(values: List[str]) -> str:
    return ",".join(values)


def _row_to_dict(r: BizEnvDetectRecord) -> Dict[str, Any]:
    inspector_ids = _split_csv(r.inspector_ids)
    inspector_names = _split_csv(r.inspector_names)
    return {
        "id": r.id,
        "recordNo": r.record_no,
        "detectTime": format_date(r.detect_time),
        "resultTime": format_date(r.result_time) if r.result_time else "",
        "inspectorIds": inspector_ids,
        "inspectorNames": inspector_names,
        "inspectorDisplay": " / ".join(inspector_names) if inspector_names else "",
        "envDetectItemId": r.env_detect_item_id,
        "itemCode": r.item_code,
        "itemName": r.item_name,
        "resultDesc": r.result_desc or "",
        "attachmentUrls": r.attachment_urls or "",
        "tenantNo": r.tenant_no,
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
    }


def _generate_record_no(db: Session, tenant_no: str) -> str:
    today_str = datetime.now().strftime("%Y%m%d")
    prefix = f"HJ{today_str}"
    existing = (
        db.query(BizEnvDetectRecord.record_no)
        .filter(
            BizEnvDetectRecord.tenant_no == tenant_no,
            BizEnvDetectRecord.record_no.like(f"{prefix}%"),
            BizEnvDetectRecord.del_flag == 0,
        )
        .order_by(BizEnvDetectRecord.record_no.desc())
        .first()
    )
    seq = 1
    if existing and existing[0]:
        try:
            seq = int(existing[0][-4:]) + 1
        except ValueError:
            seq = 1
    return f"{prefix}{seq:04d}"


def _resolve_inspector_snapshots(
    db: Session, tenant_no: str, inspector_ids: List[str]
) -> Tuple[str, str]:
    ids: List[str] = []
    seen: set = set()
    for raw in inspector_ids or []:
        uid = (raw or "").strip()
        if not uid or uid in seen:
            continue
        seen.add(uid)
        ids.append(uid)
    if not ids:
        raise ValueError("请至少选择一名检测人")
    users = (
        db.query(SysUser)
        .filter(
            SysUser.id.in_(ids),
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
        )
        .all()
    )
    user_map = {u.id: u for u in users}
    names: List[str] = []
    for uid in ids:
        user = user_map.get(uid)
        if not user:
            raise ValueError("检测人无效或不存在")
        name = (user.nick_name or user.user_no or "").strip() or uid
        names.append(name)
    return _join_csv(ids), _join_csv(names)


def _resolve_item_snapshot(
    db: Session, item_id: str, tenant_no: str
) -> Tuple[str, str, str]:
    item = item_svc.get_env_detect_item(db, item_id, tenant_no=tenant_no)
    if not item:
        raise ValueError("检测项目不存在")
    if item.status != "1":
        raise ValueError("检测项目已停用，请重新选择")
    return item.id, item.item_code, item.item_name


def get_env_detect_record(
    db: Session, row_id: str, tenant_no: Optional[str] = None
) -> Optional[BizEnvDetectRecord]:
    q = db.query(BizEnvDetectRecord).filter(
        and_(BizEnvDetectRecord.id == row_id, BizEnvDetectRecord.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizEnvDetectRecord.tenant_no == tenant_no)
    return q.first()


def list_env_detect_records(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    env_detect_item_id: Optional[str] = None,
    inspector_id: Optional[str] = None,
    detect_time_start: Optional[datetime] = None,
    detect_time_end: Optional[datetime] = None,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizEnvDetectRecord).filter(BizEnvDetectRecord.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizEnvDetectRecord.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizEnvDetectRecord.record_no.like(like),
                    BizEnvDetectRecord.inspector_names.like(like),
                    BizEnvDetectRecord.result_desc.like(like),
                )
            )
        item_id = (env_detect_item_id or "").strip()
        if item_id:
            q0 = q0.filter(BizEnvDetectRecord.env_detect_item_id == item_id)
        insp_id = (inspector_id or "").strip()
        if insp_id:
            q0 = q0.filter(func.find_in_set(insp_id, BizEnvDetectRecord.inspector_ids) > 0)
        if detect_time_start:
            q0 = q0.filter(BizEnvDetectRecord.detect_time >= detect_time_start)
        if detect_time_end:
            q0 = q0.filter(BizEnvDetectRecord.detect_time <= detect_time_end)
        return q0

    total = (
        _filtered()
        .order_by(None)
        .with_entities(func.count(BizEnvDetectRecord.id))
        .scalar()
        or 0
    )
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizEnvDetectRecord.detect_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_row_to_dict(x) for x in rows], int(total)


def create_env_detect_record(
    db: Session,
    *,
    detect_time: datetime,
    result_time: Optional[datetime] = None,
    inspector_ids: List[str],
    env_detect_item_id: str,
    result_desc: str,
    attachment_urls: Optional[str] = None,
    tenant_no: str,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizEnvDetectRecord:
    desc = (result_desc or "").strip()
    if not desc:
        raise ValueError("检测结果描述不能为空")
    if not detect_time:
        raise ValueError("检测时间不能为空")
    ids_csv, names_csv = _resolve_inspector_snapshots(db, tenant_no, inspector_ids)
    item_id, item_code, item_name = _resolve_item_snapshot(db, env_detect_item_id, tenant_no)
    obj = BizEnvDetectRecord(
        id=generate_uuid(),
        record_no=_generate_record_no(db, tenant_no),
        detect_time=detect_time,
        result_time=result_time,
        inspector_ids=ids_csv,
        inspector_names=names_csv,
        env_detect_item_id=item_id,
        item_code=item_code,
        item_name=item_name,
        result_desc=desc,
        attachment_urls=(attachment_urls or "").strip() or None,
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def update_env_detect_record(
    db: Session,
    row_id: str,
    *,
    detect_time: Optional[datetime] = None,
    result_time: Optional[datetime] = None,
    update_result_time: bool = False,
    inspector_ids: Optional[List[str]] = None,
    env_detect_item_id: Optional[str] = None,
    result_desc: Optional[str] = None,
    attachment_urls: Optional[str] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizEnvDetectRecord:
    obj = get_env_detect_record(db, row_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("登记记录不存在")
    if detect_time is not None:
        if not detect_time:
            raise ValueError("检测时间不能为空")
        obj.detect_time = detect_time
    if update_result_time:
        obj.result_time = result_time
    if inspector_ids is not None:
        ids_csv, names_csv = _resolve_inspector_snapshots(db, obj.tenant_no, inspector_ids)
        obj.inspector_ids = ids_csv
        obj.inspector_names = names_csv
    if env_detect_item_id is not None:
        item_id, item_code, item_name = _resolve_item_snapshot(
            db, env_detect_item_id, obj.tenant_no
        )
        obj.env_detect_item_id = item_id
        obj.item_code = item_code
        obj.item_name = item_name
    if result_desc is not None:
        desc = (result_desc or "").strip()
        if not desc:
            raise ValueError("检测结果描述不能为空")
        obj.result_desc = desc
    if attachment_urls is not None:
        obj.attachment_urls = (attachment_urls or "").strip() or None
    if remark is not None:
        obj.remark = remark.strip() or None
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_env_detect_record(
    db: Session, row_id: str, tenant_no: Optional[str] = None
) -> None:
    obj = get_env_detect_record(db, row_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("登记记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()