350 lines
14 KiB
Python
350 lines
14 KiB
Python
"""跨 app 共享的单据行解析：录入单位换算 + 价格。
|
||
|
||
供 sales / purchase services 复用，保证"单位换算 + 取价 + 最低售价"语义一致。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal, InvalidOperation, ROUND_DOWN
|
||
|
||
from apps.catalog.services import to_base
|
||
from apps.catalog.services import UnitConversionNotFound
|
||
|
||
|
||
class InvalidLinePrice(Exception):
|
||
"""行单价非法（非数字 / 负数 / 超上限）。负价等于倒贴出货，必须拦住。"""
|
||
|
||
def __init__(self, product_code: str, raw, reason: str):
|
||
self.product_code = product_code
|
||
self.raw = raw
|
||
self.reason = reason
|
||
super().__init__(f"invalid unit_price for {product_code}: {raw!r} ({reason})")
|
||
|
||
|
||
MAX_LINE_PRICE = Decimal("1000000000") # 10 亿
|
||
|
||
|
||
class InvalidLineQuantity(Exception):
|
||
"""行数量非法（非数字 / ≤0 / 超出上限）。
|
||
|
||
这是资损级校验：负数量会反向出库（凭空增库存），必须在此拦住。
|
||
"""
|
||
|
||
def __init__(self, product_code: str, raw, reason: str):
|
||
self.product_code = product_code
|
||
self.raw = raw
|
||
self.reason = reason
|
||
super().__init__(f"invalid quantity for {product_code}: {raw!r} ({reason})")
|
||
|
||
|
||
# 单行数量上下界
|
||
# - 上限：防止误输天文数字撑爆 Decimal 精度与库存计算
|
||
# - 下限：数量以 0.0001 为最小精度（数据库 decimal_places=4），
|
||
# 但业务上「0.0001 瓶」毫无意义——会产出 0.0016 元的荒谬单据。
|
||
# 称重类商品（kg）实际也需要 ≥0.001，这里统一取 0.0001 之上的一档。
|
||
MAX_LINE_QUANTITY = Decimal("1000000000") # 10 亿
|
||
MIN_LINE_QUANTITY = Decimal("0.001") # 0.001（称重最小实用量）
|
||
|
||
|
||
class BelowMinPrice(Exception):
|
||
"""成交价低于商品最低售价（需审批放行）。"""
|
||
|
||
def __init__(self, product_code: str, unit_price: Decimal, min_price: Decimal):
|
||
self.product_code = product_code
|
||
self.unit_price = unit_price
|
||
self.min_price = min_price
|
||
super().__init__(
|
||
f"price {unit_price} below min_sale_price {min_price} for {product_code}"
|
||
)
|
||
|
||
|
||
def resolve_unit(product, ln):
|
||
"""从行 dict 里解析录入单位（支持 Unit 实例 / pk / code）。返回 Unit 或 None。"""
|
||
unit = ln.get("source_unit") or ln.get("unit")
|
||
if unit in (None, ""):
|
||
return None
|
||
if hasattr(unit, "pk"):
|
||
return unit
|
||
from apps.catalog.models import Unit
|
||
|
||
# 先按 pk 再按 code
|
||
u = Unit.objects.filter(tenant=product.tenant, pk=unit).first()
|
||
if u is None and isinstance(unit, str):
|
||
u = Unit.objects.filter(tenant=product.tenant, code=unit).first()
|
||
return u
|
||
|
||
|
||
def parse_quantity(product, raw) -> Decimal:
|
||
"""把原始输入解析成正数量；非法一律抛 InvalidLineQuantity。"""
|
||
code = getattr(product, "code", str(product))
|
||
if raw is None or (isinstance(raw, str) and not raw.strip()):
|
||
raise InvalidLineQuantity(code, raw, "数量不能为空")
|
||
try:
|
||
qty = Decimal(str(raw))
|
||
except (InvalidOperation, ValueError, TypeError):
|
||
raise InvalidLineQuantity(code, raw, "不是合法数字")
|
||
|
||
if not qty.is_finite():
|
||
raise InvalidLineQuantity(code, raw, "数量必须是有限数")
|
||
if qty <= 0:
|
||
raise InvalidLineQuantity(code, raw, "数量必须大于 0")
|
||
if qty < MIN_LINE_QUANTITY:
|
||
raise InvalidLineQuantity(
|
||
code, raw, f"数量不能小于 {MIN_LINE_QUANTITY}（避免产生金额极小的无效单据）"
|
||
)
|
||
if qty > MAX_LINE_QUANTITY:
|
||
raise InvalidLineQuantity(code, raw, f"数量超过上限 {MAX_LINE_QUANTITY}")
|
||
return qty
|
||
|
||
|
||
def resolve_line_quantity(product, ln) -> dict:
|
||
"""解析行数量：返回 {quantity(基本单位), source_unit, source_quantity}。
|
||
|
||
数量必须为正（负数会反向出库）、在合理上限内——校验集中在 parse_quantity。
|
||
"""
|
||
if not isinstance(ln, dict):
|
||
raise InvalidLineQuantity(getattr(product, "code", "?"), ln, "行数据格式错误")
|
||
src_unit = resolve_unit(product, ln)
|
||
raw_qty = parse_quantity(product, ln.get("quantity"))
|
||
if src_unit is None:
|
||
return {"quantity": raw_qty, "source_unit": None, "source_quantity": None}
|
||
src_qty = raw_qty
|
||
qty = to_base(product, src_unit, raw_qty)
|
||
return {"quantity": qty, "source_unit": src_unit, "source_quantity": src_qty}
|
||
|
||
|
||
def parse_price(product, raw) -> Decimal:
|
||
"""把原始输入解析成合法单价（必须 > 0）。
|
||
|
||
为什么不接受 0：
|
||
- 销售 0 元 = 白送（赠品应走专门流程，不该产生 0 金额单据）
|
||
- 采购 0 元入库会**污染加权平均成本**（把真实成本拉低，导致毛利虚高）
|
||
负价同样拒绝（倒贴出货）。
|
||
"""
|
||
code = getattr(product, "code", str(product))
|
||
try:
|
||
price = Decimal(str(raw))
|
||
except (InvalidOperation, ValueError, TypeError):
|
||
raise InvalidLinePrice(code, raw, "不是合法数字")
|
||
if not price.is_finite():
|
||
raise InvalidLinePrice(code, raw, "单价必须是有限数")
|
||
if price <= 0:
|
||
raise InvalidLinePrice(code, raw, "单价必须大于 0（0 元单据无业务意义）")
|
||
if price > MAX_LINE_PRICE:
|
||
raise InvalidLinePrice(code, raw, f"单价超过上限 {MAX_LINE_PRICE}")
|
||
return price
|
||
|
||
|
||
def resolve_line_price(product, ln, *, base_price: Decimal | None = None,
|
||
rate: Decimal | None = None) -> Decimal:
|
||
"""解析行单价。
|
||
|
||
- 显式传入 unit_price → 校验后使用（按录入单位计价）
|
||
- 未传 → base_price（基本单位默认价）× rate（录入单位换算率）；
|
||
无录入单位时直接用 base_price。
|
||
"""
|
||
raw = ln.get("unit_price")
|
||
if raw not in (None, ""):
|
||
return parse_price(product, raw)
|
||
if base_price is None:
|
||
raise ValueError(f"line for {product.code} missing unit_price and no default price")
|
||
if rate is not None:
|
||
return base_price * rate
|
||
return base_price
|
||
|
||
|
||
def check_min_price(product, *, unit_price: Decimal, rate: Decimal | None = None,
|
||
allow_below_min: bool = False) -> None:
|
||
"""最低售价校验（按基本单位折算比较）。allow_below_min=True 视为审批放行。
|
||
|
||
放行是有后果的（可能亏本销售），因此记录审计——调用方拿不到 request，
|
||
这里只记业务事实，用户归属由上层补。
|
||
"""
|
||
min_price = getattr(product, "min_sale_price", Decimal("0")) or Decimal("0")
|
||
if min_price <= 0:
|
||
return
|
||
# unit_price is always expressed in the selected source unit. Convert it
|
||
# back to the product's base-unit price before comparing with min_sale_price.
|
||
# (Auto-quoting multiplies the base price by rate; multiplying again here
|
||
# would both mask under-minimum prices and make the frontend appear correct
|
||
# only by accident.)
|
||
base_price = unit_price / rate if rate is not None else unit_price
|
||
if base_price < min_price:
|
||
if not allow_below_min:
|
||
raise BelowMinPrice(product.code, unit_price, min_price)
|
||
# 审批放行 → 留痕
|
||
try:
|
||
from apps.core import audit as audit_log
|
||
|
||
audit_log.log_force_release(
|
||
tenant=product.tenant,
|
||
kind="below_min_price",
|
||
target=product.code,
|
||
detail={
|
||
"product_code": product.code,
|
||
"unit_price": str(unit_price),
|
||
"min_price": str(min_price),
|
||
"base_price": str(base_price),
|
||
},
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
ALLOWED_ROUND_TO = {Decimal("0.01"), Decimal("0.1"), Decimal("1")}
|
||
|
||
|
||
def resolve_round_to(raw) -> Decimal | None:
|
||
"""校验抹零精度：只允许 0.01 / 0.1 / 1（None 或 0 = 不抹零）。"""
|
||
if raw in (None, "", 0, "0"):
|
||
return None
|
||
try:
|
||
value = Decimal(str(raw))
|
||
except (InvalidOperation, ValueError, TypeError):
|
||
raise ValueError(f"抹零精度不合法：{raw!r}（仅支持 0.01 / 0.1 / 1）")
|
||
if value == 0:
|
||
return None
|
||
if value not in ALLOWED_ROUND_TO:
|
||
raise ValueError(f"抹零精度不合法：{raw!r}（仅支持 0.01 / 0.1 / 1）")
|
||
return value
|
||
|
||
|
||
def compute_round_off(total: Decimal, round_to: Decimal | None) -> Decimal:
|
||
"""抹零额：向下取整到 round_to 的倍数，差值即抹零金额。
|
||
|
||
例：total=100.56, round_to=1 → 0.56；round_to=0.1 → 0.06；round_to=0.01 → 0。
|
||
"""
|
||
if round_to in (None, Decimal("0")):
|
||
return Decimal("0")
|
||
round_to = Decimal(str(round_to))
|
||
if round_to <= 0:
|
||
return Decimal("0")
|
||
units = (total / round_to).to_integral_value(rounding=ROUND_DOWN)
|
||
rounded = units * round_to
|
||
return (total - rounded).quantize(Decimal("0.0001"))
|
||
|
||
|
||
# ============================================================
|
||
# 税率（批次 D4）
|
||
# ============================================================
|
||
|
||
class InvalidLineAmount(Exception):
|
||
"""行金额非法（≤0）。数量与单价校验通过后仍可能合计为 0（如单价 0 的赠品需显式标注）。"""
|
||
|
||
def __init__(self, product_code: str, amount):
|
||
self.product_code = product_code
|
||
self.amount = amount
|
||
super().__init__(f"line amount must be > 0 for {product_code}: {amount}")
|
||
|
||
|
||
def compute_tax(amount: Decimal, tax_rate: Decimal | None) -> dict:
|
||
"""价内税拆分：amount 视为**含税**金额，拆出不含税净额与税额。
|
||
|
||
采用价内口径（国内零售/批发单据惯例：报价即含税价）：
|
||
net = amount / (1 + rate)
|
||
tax = amount - net
|
||
|
||
返回 {tax_rate, gross, net, tax}。rate 为 0/None 时 tax=0、net=gross。
|
||
"""
|
||
gross = Decimal(str(amount or 0))
|
||
rate = Decimal(str(tax_rate or 0))
|
||
if rate <= 0:
|
||
return {
|
||
"tax_rate": Decimal("0"),
|
||
"gross": gross.quantize(Decimal("0.0001")),
|
||
"net": gross.quantize(Decimal("0.0001")),
|
||
"tax": Decimal("0.0000"),
|
||
}
|
||
net = (gross / (Decimal("1") + rate)).quantize(Decimal("0.0001"))
|
||
tax = (gross - net).quantize(Decimal("0.0001"))
|
||
return {"tax_rate": rate, "gross": gross, "net": net, "tax": tax}
|
||
|
||
|
||
def resolve_tax_rate(product, line: dict | None = None) -> Decimal:
|
||
"""行税率：行显式指定 > 商品默认税率 > 0（不启用）。"""
|
||
line = line or {}
|
||
raw = line.get("tax_rate")
|
||
if raw not in (None, ""):
|
||
return Decimal(str(raw))
|
||
return Decimal(str(getattr(product, "tax_rate", 0) or 0))
|
||
|
||
|
||
# ============================================================
|
||
# 单号生成（迭代第 4 轮：并发安全 + 删除安全）
|
||
# ============================================================
|
||
|
||
import re as _re
|
||
|
||
_BILL_NO_RE = _re.compile(r"^([A-Z]+)(\d{8})(\d+)$")
|
||
|
||
# 生成单号时的最大重试次数（并发冲突时重试）
|
||
BILL_NO_MAX_ATTEMPTS = 20
|
||
|
||
|
||
def next_bill_no(tenant, prefix: str, model, *, date_str: str | None = None,
|
||
field: str = "bill_no") -> str:
|
||
"""按 `前缀+日期+4位序号` 生成单号。
|
||
|
||
相比旧实现（`count() + 1`）修了两个问题：
|
||
1. **并发撞号**：多个事务同时 `count()` 会拿到同一个数 → 生成相同单号
|
||
（PG 实测：9 个并发建单产生重复的 `XS202609110001`）。
|
||
2. **删除后重复**：`count()` 在删单后会变小，重新生成的号可能已被用过。
|
||
例如建了 001/002 后删掉 001，count=1，下一张又是 002 → 撞唯一约束。
|
||
|
||
改用 `MAX(序号) + 1`，配合调用方的唯一约束冲突重试，
|
||
保证"不重复、不跳号（除了冲突重试）"。
|
||
|
||
注意：**本函数只负责取号，不保证唯一**。真正保证唯一的是
|
||
数据库唯一约束 + 调用方的重试（见 `create_with_unique_bill_no`）。
|
||
"""
|
||
from django.db.models import Max
|
||
|
||
day = date_str or date.today().strftime("%Y%m%d")
|
||
head = f"{prefix}{day}"
|
||
|
||
# 取该前缀+日期下已有单号的最大序号
|
||
existing = (
|
||
model.objects.filter(tenant=tenant, **{f"{field}__startswith": head})
|
||
.values_list(field, flat=True)
|
||
)
|
||
max_seq = 0
|
||
for no in existing:
|
||
m = _BILL_NO_RE.match(no)
|
||
if m and m.group(2) == day:
|
||
try:
|
||
max_seq = max(max_seq, int(m.group(3)))
|
||
except ValueError:
|
||
continue
|
||
return f"{head}{max_seq + 1:04d}"
|
||
|
||
|
||
def create_with_unique_bill_no(model, *, tenant, prefix: str, defaults: dict,
|
||
field: str = "bill_no",
|
||
attempts: int = BILL_NO_MAX_ATTEMPTS):
|
||
"""创建一条带唯一单号的记录，冲突时自动重试。
|
||
|
||
用**乐观并发控制**：不依赖锁（`select_for_update` 锁不住不存在的行——
|
||
这是上一轮修库存竞态时学到的教训），而是"先试、撞了再换号重试"。
|
||
|
||
`defaults` 为除 bill_no 外的创建参数。
|
||
"""
|
||
from django.db import IntegrityError, transaction
|
||
|
||
last_exc = None
|
||
for _ in range(attempts):
|
||
bill_no = next_bill_no(tenant, prefix, model, field=field)
|
||
try:
|
||
# 保存点：冲突只回滚这一次尝试，不污染外层事务
|
||
with transaction.atomic():
|
||
return model.objects.create(
|
||
tenant=tenant, **{field: bill_no}, **defaults
|
||
)
|
||
except IntegrityError as exc:
|
||
last_exc = exc
|
||
continue
|
||
|
||
raise RuntimeError(
|
||
f"生成唯一单号失败（尝试 {attempts} 次）：{prefix} · {last_exc}"
|
||
)
|