zkxw/backend/app/crud/sample_customer_service.py

"""样本客户服务扩展 CRUD。"""
from __future__ import annotations

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

from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud.customer_service_dict import ensure_dict_value
from app.models.models import BizSampleCustomerService
from app.services.sample_customer_service_register import (
    CUSTOMER_SERVICE_FIELD_KEYS,
    merge_customer_service_fields,
    normalize_dict_value,
)

DICT_TYPE_BY_FIELD = {
    "sms_send_status": "sms_send_status",
    "microbe_followup_status": "microbe_followup_status",
    "fullmoon_followup_status": "fullmoon_followup_status",
    "household_registered": "household_registered",
}


def _format_date(value: Optional[date]) -> Optional[str]:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    return value.strftime("%Y-%m-%d")


def _format_datetime(value: Optional[datetime]) -> Optional[str]:
    if value is None:
        return None
    return value.strftime("%Y-%m-%d %H:%M:%S")


def _row_field_dict(row: BizSampleCustomerService) -> Dict[str, Any]:
    return {key: getattr(row, key) for key in CUSTOMER_SERVICE_FIELD_KEYS}


def get_by_sample_code(
    db: Session, tenant_no: str, sample_code: str
) -> Optional[BizSampleCustomerService]:
    code = (sample_code or "").strip()
    if not code:
        return None
    return (
        db.query(BizSampleCustomerService)
        .filter(
            BizSampleCustomerService.tenant_no == tenant_no,
            BizSampleCustomerService.sample_code == code,
            BizSampleCustomerService.del_flag == 0,
        )
        .first()
    )


def get_map_by_sample_codes(
    db: Session, tenant_no: str, sample_codes: List[str]
) -> Dict[str, BizSampleCustomerService]:
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not codes:
        return {}
    rows = (
        db.query(BizSampleCustomerService)
        .filter(
            BizSampleCustomerService.tenant_no == tenant_no,
            BizSampleCustomerService.sample_code.in_(codes),
            BizSampleCustomerService.del_flag == 0,
        )
        .all()
    )
    return {row.sample_code: row for row in rows}


def row_to_dict(row: BizSampleCustomerService) -> Dict[str, Any]:
    return {
        "id": row.id,
        "sampleCode": row.sample_code,
        "collectionProtocolId": row.collection_protocol_id,
        "smsSendDate": _format_date(row.sms_send_date),
        "smsSendStatus": row.sms_send_status,
        "microbeFollowupDate": _format_date(row.microbe_followup_date),
        "microbeFollowupStatus": row.microbe_followup_status,
        "fullmoonFollowupDate": _format_date(row.fullmoon_followup_date),
        "fullmoonFollowupStatus": row.fullmoon_followup_status,
        "householdRegistered": row.household_registered,
        "einvoiceSendTime": _format_datetime(row.einvoice_send_time),
        "effectiveAddress": row.effective_address,
    }


def upsert_by_sample_code(
    db: Session,
    *,
    tenant_no: str,
    sample_code: str,
    collection_protocol_id: Optional[str],
    user_id: Optional[str],
    incoming: Dict[str, Any],
) -> BizSampleCustomerService:
    code = (sample_code or "").strip()
    if not code:
        raise ValueError("sample_code is required")

    row = get_by_sample_code(db, tenant_no, code)
    now = datetime.now()
    if row is None:
        row = BizSampleCustomerService(
            id=generate_uuid(),
            sample_code=code,
            collection_protocol_id=collection_protocol_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,
        )
        db.add(row)
    else:
        if collection_protocol_id and not row.collection_protocol_id:
            row.collection_protocol_id = collection_protocol_id
        row.update_by = user_id
        row.update_time = now

    existing = _row_field_dict(row)
    merged = merge_customer_service_fields(existing, incoming)

    for field, dict_type in DICT_TYPE_BY_FIELD.items():
        if field not in merged:
            continue
        value = merged[field]
        if value is None or normalize_dict_value(value) is None:
            continue
        merged[field] = ensure_dict_value(
            db, tenant_no, dict_type, value, user_id=user_id
        )

    for key in CUSTOMER_SERVICE_FIELD_KEYS:
        if key in merged:
            setattr(row, key, merged[key])

    db.flush()
    return row