shanxia2.0/backend/app/services/db/mappers.py

"""地磅 ORM → API Read 模型映射。"""
from __future__ import annotations

import json
from datetime import date, datetime, time

from app.core.utils import format_datetime

# 三方预生成出入库单「日期未填」占位（原系统 date=""，DDL 为 NOT NULL DATE）
EMPTY_INVENTORY_DATE = date(1970, 1, 1)
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
from app.schemas.db.inventory import InboundOutboundRead
from app.schemas.db.plans import PurchasePlanRead, PurchasePlanWithFillLink
from app.schemas.db.weighing import WeighingSlipRead


def fmt_date(value: date | datetime | str | None) -> str | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    if isinstance(value, date):
        return value.strftime("%Y-%m-%d")
    return str(value)


def fmt_datetime(value: datetime | str | None) -> str | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return format_datetime(value)
    return str(value)


def plan_to_read(row: DbPurchasePlan, *, executed_quantity: float | None = None) -> PurchasePlanRead:
    return PurchasePlanRead(
        id=row.id,
        planNo=row.plan_no,
        planDate=fmt_date(row.plan_date) or "",
        startDate=fmt_date(row.start_date),
        endDate=fmt_date(row.end_date),
        businessType=row.business_type,
        productId=row.product_id,
        product=row.product,
        specification=row.specification,
        plannedQuantity=row.planned_quantity,
        vehicleNo=row.vehicle_no or "",
        customerName=row.customer_name,
        remarks=row.remark,
        executionStatus=row.execution_status,
        executedAt=fmt_datetime(row.executed_at),
        lastWeighingSlipId=row.last_weighing_slip_id,
        executedQuantity=(
            executed_quantity if executed_quantity is not None else (row.executed_quantity or 0)
        ),
        autoGenerateEnabled=row.auto_generate_enabled or 0,
        autoGenerateEndMonth=row.auto_generate_end_month,
        autoGenerateLastMonth=row.auto_generate_last_month,
        isDeleted=row.del_flag,
        createdAt=format_datetime(row.create_time) or "",
    )


def plan_to_read_with_fill_link(
    row: DbPurchasePlan,
    *,
    executed_quantity: float | None = None,
    fill_link_inventory_id: str | None = None,
) -> PurchasePlanWithFillLink:
    base = plan_to_read(row, executed_quantity=executed_quantity)
    return PurchasePlanWithFillLink(
        **base.model_dump(by_alias=False),
        fill_link_inventory_id=fill_link_inventory_id,
    )


def is_empty_inventory_date(value: date | None) -> bool:
    return value is None or value == EMPTY_INVENTORY_DATE


def inventory_date_to_api(value: date | None) -> str:
    if is_empty_inventory_date(value):
        return ""
    return fmt_date(value) or ""


def _parse_site_feedback_images(raw: str | None) -> list[str] | None:
    if not raw:
        return None
    try:
        parsed = json.loads(raw)
        if isinstance(parsed, list):
            items = [str(x) for x in parsed if x]
            return items or None
    except (TypeError, json.JSONDecodeError):
        pass
    items = [part.strip() for part in raw.split(",") if part.strip()]
    return items or None


def inventory_to_read(
    row: DbInboundOutbound,
    *,
    weighing: DbWeighingSlip | None = None,
) -> InboundOutboundRead:
    gross_weight = None
    tare_weight = None
    deduction = None
    net_weight = None
    gross_weight_time = None
    tare_weight_time = None
    has_weighing_slip = False
    if weighing:
        gross_weight = weighing.gross_weight
        tare_weight = weighing.tare_weight
        deduction = weighing.deduction
        net_weight = weighing.net_weight
        gross_weight_time = fmt_datetime(weighing.gross_weight_time)
        tare_time = fmt_datetime(weighing.tare_weight_time)
        tare_weight_time = tare_time if tare_time and tare_time.strip() else None
        has_weighing_slip = True

    return InboundOutboundRead(
        id=row.id,
        date=inventory_date_to_api(row.date),
        partnerCompany=row.partner_company,
        businessType=row.business_type,
        productId=row.product_id,
        product=row.product,
        specification=row.specification,
        quantity=row.quantity,
        planNo=row.plan_no,
        batchNo=row.batch_no,
        vehicleNo=row.vehicle_no,
        status=row.status,
        remarks=row.remark,
        purchasePlanId=row.purchase_plan_id,
        isDeleted=row.del_flag,
        createdAt=format_datetime(row.create_time) or "",
        grossWeight=gross_weight,
        tareWeight=tare_weight,
        deduction=deduction,
        netWeight=net_weight,
        grossWeightTime=gross_weight_time,
        tareWeightTime=tare_weight_time,
        hasWeighingSlip=has_weighing_slip,
        siteFeedbackRemark=row.site_feedback_remark,
        siteFeedbackImages=_parse_site_feedback_images(row.site_feedback_images),
        siteFeedbackTime=format_datetime(row.site_feedback_time) or None,
    )


def weighing_to_read(row: DbWeighingSlip, *, inventory_date: str | None = None) -> WeighingSlipRead:
    return WeighingSlipRead(
        id=row.id,
        vehicleNo=row.vehicle_no,
        planNo=row.plan_no,
        productId=row.product_id,
        cargoName=row.cargo_name,
        specification=row.specification,
        businessType=row.business_type,
        batchNo=row.batch_no,
        supplier=row.supplier,
        consignee=row.consignee,
        grossWeight=row.gross_weight,
        grossWeightTime=fmt_datetime(row.gross_weight_time) or "",
        tareWeight=row.tare_weight,
        tareWeightTime=fmt_datetime(row.tare_weight_time) or "",
        deduction=row.deduction,
        netWeight=row.net_weight,
        plannedQuantity=row.planned_quantity,
        executedQuantity=row.executed_quantity,
        purchasePlanId=row.purchase_plan_id,
        inboundOutboundId=row.inbound_outbound_id,
        weighingPoint=row.weighing_point,
        remarks=row.remark,
        currentQuantity=row.current_quantity,
        isVoided=row.is_voided,
        createdAt=format_datetime(row.create_time) or "",
        inventoryDate=inventory_date,
    )


def datetime_end_of_day(value: date) -> datetime:
    return datetime.combine(value, time(23, 59, 59))


def parse_date_str(value: str | None) -> date | None:
    if not value:
        return None
    try:
        return datetime.strptime(value[:10], "%Y-%m-%d").date()
    except ValueError:
        return None


def parse_datetime_str(value: str | None) -> datetime | None:
    if not value:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(value[:19] if fmt != "%Y-%m-%d" else value[:10], fmt)
        except ValueError:
            continue
    return None