206 lines
6.8 KiB
Python
206 lines
6.8 KiB
Python
"""地磅 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
|