zkxw/backend/app/crud/protocol_operation_log.py

"""协议单操作过程日志（Issue #24）。

记录细胞存储 / 科研捐赠协议在 Web、小程序端的 create/update/delete/状态类动作，
保存操作前、动作、操作后快照，供责任归属追溯。append-only，不 commit。
"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import BizCollectionProtocol, BizProtocolOperationLog, SysUser

BIZ_CELL = "cell_storage"
BIZ_DONATION = "research_donation"

ACTION_CREATE = "create"
ACTION_UPDATE = "update"
ACTION_DELETE = "delete"
ACTION_VOID = "void"
ACTION_ACTIVATE = "activate"
ACTION_EXPIRE = "expire"
ACTION_TO_RESEARCH = "to_research"
ACTION_TO_STORAGE = "to_storage"
ACTION_AUDIT = "audit"

CHANNEL_WEB = "web"
CHANNEL_MINI = "mini"

_STATUS_ACTION_MAP = {
    "sxz": ACTION_ACTIVATE,
    "ygq": ACTION_EXPIRE,
    "yzf": ACTION_VOID,
    "zyk": ACTION_TO_RESEARCH,
    "zcck": ACTION_TO_STORAGE,
}


def _json_dumps(payload: Optional[Dict[str, Any]]) -> Optional[str]:
    if payload is None:
        return None
    return json.dumps(payload, ensure_ascii=False, default=str, separators=(",", ":"))


def _json_loads(raw: Optional[str]) -> Optional[Dict[str, Any]]:
    if not raw:
        return None
    try:
        value = json.loads(raw)
    except (TypeError, ValueError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None


def resolve_action_from_status(
    *,
    default_action: str,
    before_status: Optional[str] = None,
    after_status: Optional[str] = None,
) -> str:
    """仅对 update 默认动作按状态流转细化；显式动作（create/delete/audit 等）保持原值。"""
    action = (default_action or "").strip() or ACTION_UPDATE
    if action != ACTION_UPDATE:
        return action
    before = (before_status or "").strip()
    after = (after_status or "").strip()
    if after and after != before and after in _STATUS_ACTION_MAP:
        return _STATUS_ACTION_MAP[after]
    return action


def _operator_name(db: Session, user_id: Optional[str]) -> str:
    uid = (user_id or "").strip()
    if not uid:
        return ""
    row = db.query(SysUser.nick_name).filter(SysUser.id == uid).first()
    if not row:
        return ""
    return (row[0] or "").strip()


def snapshot_cell_protocol(
    db: Session,
    *,
    protocol_id: str,
    tenant_no: str,
    include_deleted: bool = False,
) -> Optional[Dict[str, Any]]:
    """采集协议主表 + 当事人 + 采集信息 + 样本快照。"""
    from app.crud import collection_protocol as cell_crud

    pid = (protocol_id or "").strip()
    if not pid:
        return None
    q = db.query(BizCollectionProtocol).filter(
        BizCollectionProtocol.id == pid,
        BizCollectionProtocol.tenant_no == tenant_no,
    )
    if not include_deleted:
        q = q.filter(BizCollectionProtocol.del_flag == 0)
    row = q.first()
    if not row:
        return None
    party = cell_crud.get_party_customer(db, pid, tenant_no)
    collection = cell_crud.get_collection_info(db, pid, tenant_no)
    samples = cell_crud.list_samples(db, pid, tenant_no)
    return {
        "protocol": cell_crud._protocol_to_dict(row, db),
        "partyCustomer": cell_crud._party_customer_to_dict(party) if party else None,
        "collectionInfo": cell_crud._collection_info_to_dict(collection)
        if collection
        else None,
        "samples": [cell_crud._sample_to_dict(s) for s in samples],
    }


def snapshot_donation(
    db: Session,
    *,
    donation_id: str,
    tenant_no: str,
    include_deleted: bool = False,
) -> Optional[Dict[str, Any]]:
    """科研捐赠主表 + 当事人 + 采集信息 + 样本快照。"""
    from app.crud import research_donation as donation_crud
    from app.models.research_donation import BizResearchDonation

    did = (donation_id or "").strip()
    if not did:
        return None
    q = db.query(BizResearchDonation).filter(
        BizResearchDonation.id == did,
        BizResearchDonation.tenant_no == tenant_no,
    )
    if not include_deleted:
        q = q.filter(BizResearchDonation.del_flag == 0)
    row = q.first()
    if not row:
        return None
    party = donation_crud.get_party(db, did, tenant_no)
    collection = donation_crud.get_collection(db, did, tenant_no)
    samples = donation_crud.list_samples(db, did, tenant_no)
    return {
        "donation": donation_crud._donation_to_dict(row, db, tenant_no),
        "party": donation_crud._party_to_dict(party) if party else None,
        "collection": donation_crud._collection_to_dict(collection)
        if collection
        else None,
        "samples": [donation_crud._sample_to_dict(s) for s in samples],
    }


def _protocol_no_from_snapshot(
    biz_type: str, snapshot: Optional[Dict[str, Any]]
) -> Optional[str]:
    if not snapshot:
        return None
    if biz_type == BIZ_CELL:
        protocol = snapshot.get("protocol") or {}
        return (protocol.get("protocolNo") or "").strip() or None
    donation = snapshot.get("donation") or {}
    return (donation.get("donationNo") or "").strip() or None


def _status_from_snapshot(
    biz_type: str, snapshot: Optional[Dict[str, Any]]
) -> Optional[str]:
    if not snapshot:
        return None
    if biz_type == BIZ_CELL:
        protocol = snapshot.get("protocol") or {}
        return (protocol.get("protocolStatus") or "").strip() or None
    donation = snapshot.get("donation") or {}
    return (donation.get("donationStatus") or "").strip() or None


def append_protocol_operation_log(
    db: Session,
    *,
    biz_type: str,
    biz_id: str,
    action: str,
    client_channel: str = CHANNEL_WEB,
    before: Optional[Dict[str, Any]] = None,
    after: Optional[Dict[str, Any]] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    operator_user_name: Optional[str] = None,
    operate_time: Optional[datetime] = None,
    remark: Optional[str] = None,
    create_dept_id: str = "",
) -> BizProtocolOperationLog:
    """追加一条操作日志（不 commit）。"""
    now = operate_time or datetime.now()
    name = (operator_user_name or "").strip() or _operator_name(db, user_id)
    protocol_no = _protocol_no_from_snapshot(biz_type, after) or _protocol_no_from_snapshot(
        biz_type, before
    )
    obj = BizProtocolOperationLog(
        id=generate_uuid(),
        biz_type=(biz_type or "").strip(),
        biz_id=(biz_id or "").strip(),
        protocol_no=protocol_no,
        action=(action or "").strip() or ACTION_UPDATE,
        client_channel=(client_channel or CHANNEL_WEB).strip() or CHANNEL_WEB,
        before_json=_json_dumps(before),
        after_json=_json_dumps(after),
        operator_user_id=(user_id or "").strip() or None,
        operator_user_name=name or None,
        operate_time=now,
        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=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    return obj


def record_cell_operation(
    db: Session,
    *,
    protocol_id: str,
    action: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    client_channel: str = CHANNEL_WEB,
    before: Optional[Dict[str, Any]] = None,
    after: Optional[Dict[str, Any]] = None,
    remark: Optional[str] = None,
    create_dept_id: str = "",
) -> Optional[BizProtocolOperationLog]:
    """记录细胞存储协议操作；after 缺省时按当前库态补齐。"""
    pid = (protocol_id or "").strip()
    if not pid:
        return None
    after_snap = after
    if after_snap is None and action != ACTION_DELETE:
        after_snap = snapshot_cell_protocol(db, protocol_id=pid, tenant_no=tenant_no)
    resolved = resolve_action_from_status(
        default_action=action,
        before_status=_status_from_snapshot(BIZ_CELL, before),
        after_status=_status_from_snapshot(BIZ_CELL, after_snap),
    )
    return append_protocol_operation_log(
        db,
        biz_type=BIZ_CELL,
        biz_id=pid,
        action=resolved,
        client_channel=client_channel,
        before=before,
        after=after_snap,
        tenant_no=tenant_no,
        user_id=user_id,
        remark=remark,
        create_dept_id=create_dept_id,
    )


def record_donation_operation(
    db: Session,
    *,
    donation_id: str,
    action: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    client_channel: str = CHANNEL_WEB,
    before: Optional[Dict[str, Any]] = None,
    after: Optional[Dict[str, Any]] = None,
    remark: Optional[str] = None,
    create_dept_id: str = "",
) -> Optional[BizProtocolOperationLog]:
    """记录科研捐赠协议操作；after 缺省时按当前库态补齐。"""
    did = (donation_id or "").strip()
    if not did:
        return None
    after_snap = after
    if after_snap is None and action != ACTION_DELETE:
        after_snap = snapshot_donation(db, donation_id=did, tenant_no=tenant_no)
    resolved = resolve_action_from_status(
        default_action=action,
        before_status=_status_from_snapshot(BIZ_DONATION, before),
        after_status=_status_from_snapshot(BIZ_DONATION, after_snap),
    )
    return append_protocol_operation_log(
        db,
        biz_type=BIZ_DONATION,
        biz_id=did,
        action=resolved,
        client_channel=client_channel,
        before=before,
        after=after_snap,
        tenant_no=tenant_no,
        user_id=user_id,
        remark=remark,
        create_dept_id=create_dept_id,
    )


def _row_to_dict(row: BizProtocolOperationLog) -> Dict[str, Any]:
    return {
        "id": row.id,
        "bizType": row.biz_type,
        "bizId": row.biz_id,
        "protocolNo": row.protocol_no or "",
        "action": row.action,
        "clientChannel": row.client_channel or CHANNEL_WEB,
        "before": _json_loads(row.before_json),
        "after": _json_loads(row.after_json),
        "operatorUserId": row.operator_user_id or "",
        "operatorUserName": row.operator_user_name or "",
        "operateTime": row.operate_time.strftime("%Y-%m-%d %H:%M:%S")
        if row.operate_time
        else "",
        "tenantNo": row.tenant_no,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S")
        if row.create_time
        else "",
    }


def list_protocol_operation_logs(
    db: Session,
    *,
    biz_type: str,
    biz_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizProtocolOperationLog)
        .filter(
            BizProtocolOperationLog.biz_type == (biz_type or "").strip(),
            BizProtocolOperationLog.biz_id == (biz_id or "").strip(),
            BizProtocolOperationLog.tenant_no == tenant_no,
            BizProtocolOperationLog.del_flag == 0,
        )
        .order_by(
            BizProtocolOperationLog.operate_time.asc(),
            BizProtocolOperationLog.create_time.asc(),
        )
        .all()
    )
    return [_row_to_dict(r) for r in rows]