490 lines
16 KiB
Python
490 lines
16 KiB
Python
import re
|
||
from datetime import datetime
|
||
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models.attachment import (
|
||
AttachmentCategory,
|
||
FinishedGoodsAttachment,
|
||
FinishedGoodsAttachmentItem,
|
||
FinishedGoodsAttachmentLocation,
|
||
FinishedGoodsBoxAttachmentItem,
|
||
)
|
||
from app.models.finished_goods import FinishedGoodsBox
|
||
from app.schemas.attachment import AttachmentConfigRequest
|
||
from app.services.box_service import (
|
||
InvalidQuantityError,
|
||
InvalidZongpaiError,
|
||
query_erp_info,
|
||
)
|
||
|
||
ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
|
||
|
||
|
||
class CategoryNotFoundError(Exception):
|
||
pass
|
||
|
||
|
||
class AttachmentItemNotFoundError(Exception):
|
||
pass
|
||
|
||
|
||
class DuplicateAttachmentLocationError(Exception):
|
||
def __init__(self, location_code: str, registered_at: str):
|
||
self.location_code = location_code
|
||
self.registered_at = registered_at
|
||
|
||
|
||
class AlreadyOffShelfAttachmentError(Exception):
|
||
def __init__(self, location_code: str, registered_at: str):
|
||
self.location_code = location_code
|
||
self.registered_at = registered_at
|
||
|
||
|
||
class DuplicateAttachmentBoxItemError(Exception):
|
||
def __init__(self, paichan_no: str, box_no: int):
|
||
self.paichan_no = paichan_no
|
||
self.box_no = box_no
|
||
|
||
|
||
class CategoryHasBoxesError(Exception):
|
||
"""配置中删除已有装箱记录的类型时抛出。"""
|
||
|
||
def __init__(self, category_name: str):
|
||
self.category_name = category_name
|
||
|
||
|
||
def _validate_zongpai(zongpai_no: str) -> str:
|
||
zongpai_no = zongpai_no.strip().upper()
|
||
if not ZONGPAI_PATTERN.match(zongpai_no):
|
||
raise InvalidZongpaiError()
|
||
return zongpai_no
|
||
|
||
|
||
def _boxed_qty_map(db: Session, zongpai_no: str) -> dict[int, int]:
|
||
"""按 category_id 汇总已装数量。"""
|
||
rows = db.execute(
|
||
select(
|
||
FinishedGoodsBoxAttachmentItem.category_id,
|
||
func.sum(FinishedGoodsBoxAttachmentItem.quantity),
|
||
)
|
||
.where(FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no)
|
||
.group_by(FinishedGoodsBoxAttachmentItem.category_id)
|
||
).all()
|
||
return {row[0]: int(row[1] or 0) for row in rows}
|
||
|
||
|
||
def _location_map(db: Session, zongpai_no: str) -> dict[int, str]:
|
||
rows = (
|
||
db.query(FinishedGoodsAttachmentLocation)
|
||
.filter(FinishedGoodsAttachmentLocation.zongpai_no == zongpai_no)
|
||
.all()
|
||
)
|
||
return {r.category_id: r.location_code for r in rows}
|
||
|
||
|
||
def _category_name_map(db: Session, category_ids: list[int]) -> dict[int, str]:
|
||
if not category_ids:
|
||
return {}
|
||
rows = (
|
||
db.query(AttachmentCategory)
|
||
.filter(AttachmentCategory.id.in_(category_ids))
|
||
.all()
|
||
)
|
||
return {r.id: r.name for r in rows}
|
||
|
||
|
||
def get_attachment_status(db: Session, zongpai_no: str) -> dict:
|
||
zongpai_no = _validate_zongpai(zongpai_no)
|
||
|
||
att = (
|
||
db.query(FinishedGoodsAttachment)
|
||
.filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
|
||
.first()
|
||
)
|
||
determination = att.determination if att else "undetermined"
|
||
|
||
items_db = (
|
||
db.query(FinishedGoodsAttachmentItem)
|
||
.filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no)
|
||
.all()
|
||
)
|
||
category_ids = [i.category_id for i in items_db]
|
||
names = _category_name_map(db, category_ids)
|
||
boxed = _boxed_qty_map(db, zongpai_no)
|
||
locations = _location_map(db, zongpai_no)
|
||
|
||
items = []
|
||
for it in items_db:
|
||
boxed_qty = boxed.get(it.category_id, 0)
|
||
items.append(
|
||
{
|
||
"category_id": it.category_id,
|
||
"name": names.get(it.category_id, "(已删除类型)"),
|
||
"expected_qty": it.expected_qty,
|
||
"boxed_qty": boxed_qty,
|
||
"location_code": locations.get(it.category_id),
|
||
"complete": boxed_qty >= it.expected_qty,
|
||
}
|
||
)
|
||
|
||
items.sort(key=lambda x: x["category_id"])
|
||
all_complete = determination == "has" and bool(items) and all(i["complete"] for i in items)
|
||
return {
|
||
"zongpai_no": zongpai_no,
|
||
"determination": determination,
|
||
"all_complete": all_complete,
|
||
"items": items,
|
||
}
|
||
|
||
|
||
def _ensure_categories_exist(db: Session, category_ids: list[int]) -> dict[int, str]:
|
||
names = _category_name_map(db, category_ids)
|
||
missing = [c for c in category_ids if c not in names]
|
||
if missing:
|
||
raise CategoryNotFoundError(missing[0])
|
||
return names
|
||
|
||
|
||
def configure_attachment(db: Session, req: AttachmentConfigRequest) -> dict:
|
||
zongpai_no = _validate_zongpai(req.zongpai_no)
|
||
|
||
# Validate categories referenced in the submitted list
|
||
submitted_ids = [it.category_id for it in req.items]
|
||
_ensure_categories_exist(db, submitted_ids)
|
||
|
||
# Upsert order-level determination row
|
||
att = (
|
||
db.query(FinishedGoodsAttachment)
|
||
.filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
|
||
.first()
|
||
)
|
||
if att is None:
|
||
att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination=req.determination)
|
||
db.add(att)
|
||
else:
|
||
att.determination = req.determination
|
||
att.updated_at = datetime.now()
|
||
db.flush()
|
||
|
||
# Load existing items
|
||
existing = (
|
||
db.query(FinishedGoodsAttachmentItem)
|
||
.filter(FinishedGoodsAttachmentItem.zongpai_no == zongpai_no)
|
||
.all()
|
||
)
|
||
existing_by_cat = {it.category_id: it for it in existing}
|
||
submitted_set = set(submitted_ids)
|
||
boxed = _boxed_qty_map(db, zongpai_no)
|
||
names = _category_name_map(db, list(existing_by_cat.keys()) + submitted_ids)
|
||
|
||
# Remove items not in the submitted list; block those with boxed > 0
|
||
for cat_id, it in existing_by_cat.items():
|
||
if cat_id not in submitted_set:
|
||
if boxed.get(cat_id, 0) > 0:
|
||
raise CategoryHasBoxesError(names.get(cat_id, str(cat_id)))
|
||
db.delete(it)
|
||
|
||
# Upsert submitted items
|
||
for submitted in req.items:
|
||
it = existing_by_cat.get(submitted.category_id)
|
||
if it is None:
|
||
db.add(
|
||
FinishedGoodsAttachmentItem(
|
||
zongpai_no=zongpai_no,
|
||
category_id=submitted.category_id,
|
||
expected_qty=submitted.expected_qty,
|
||
)
|
||
)
|
||
else:
|
||
it.expected_qty = submitted.expected_qty
|
||
it.updated_at = datetime.now()
|
||
|
||
db.commit()
|
||
return get_attachment_status(db, zongpai_no)
|
||
|
||
|
||
def _is_transit_location(location_code: str) -> bool:
|
||
return location_code.startswith("TRANS-")
|
||
|
||
|
||
def _is_temp_storage_location(location_code: str) -> bool:
|
||
return location_code.startswith("B")
|
||
|
||
|
||
def register_attachment_location(db: Session, req) -> FinishedGoodsAttachmentLocation:
|
||
zongpai_no = _validate_zongpai(req.zongpai_no)
|
||
|
||
# Category must exist
|
||
if not _category_name_map(db, [req.category_id]):
|
||
raise CategoryNotFoundError(req.category_id)
|
||
|
||
existing = (
|
||
db.query(FinishedGoodsAttachmentLocation)
|
||
.filter(
|
||
FinishedGoodsAttachmentLocation.zongpai_no == zongpai_no,
|
||
FinishedGoodsAttachmentLocation.category_id == req.category_id,
|
||
)
|
||
.first()
|
||
)
|
||
if existing:
|
||
if _is_transit_location(existing.location_code):
|
||
raise AlreadyOffShelfAttachmentError(
|
||
location_code=existing.location_code,
|
||
registered_at=existing.created_at.isoformat(),
|
||
)
|
||
|
||
target_is_transit = _is_transit_location(req.location_code)
|
||
target_is_temp = _is_temp_storage_location(req.location_code)
|
||
current_is_temp = _is_temp_storage_location(existing.location_code)
|
||
|
||
if target_is_transit or (not current_is_temp and target_is_temp):
|
||
previous_location = existing.location_code
|
||
existing.location_code = req.location_code
|
||
existing.created_at = datetime.now()
|
||
db.commit()
|
||
db.refresh(existing)
|
||
existing.previous_location = previous_location
|
||
return existing
|
||
|
||
raise DuplicateAttachmentLocationError(
|
||
location_code=existing.location_code,
|
||
registered_at=existing.created_at.isoformat(),
|
||
)
|
||
|
||
record = FinishedGoodsAttachmentLocation(
|
||
zongpai_no=zongpai_no,
|
||
category_id=req.category_id,
|
||
location_code=req.location_code,
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
db.refresh(record)
|
||
return record
|
||
|
||
|
||
def _boxed_for_category(
|
||
db: Session, zongpai_no: str, category_id: int, exclude_item_id: int | None = None
|
||
) -> int:
|
||
q = db.query(FinishedGoodsBoxAttachmentItem).filter(
|
||
FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
|
||
FinishedGoodsBoxAttachmentItem.category_id == category_id,
|
||
)
|
||
if exclude_item_id is not None:
|
||
q = q.filter(FinishedGoodsBoxAttachmentItem.id != exclude_item_id)
|
||
return sum(int(r.quantity or 0) for r in q.all())
|
||
|
||
|
||
def _ensure_attachment_item(db: Session, zongpai_no: str, category_id: int, erp_qty: int):
|
||
"""实物证据：不存在则建 item（expected 兜底 = ERP 数量），并保证 determination=has。"""
|
||
if not _category_name_map(db, [category_id]):
|
||
raise CategoryNotFoundError(category_id)
|
||
|
||
att = (
|
||
db.query(FinishedGoodsAttachment)
|
||
.filter(FinishedGoodsAttachment.zongpai_no == zongpai_no)
|
||
.first()
|
||
)
|
||
if att is None:
|
||
att = FinishedGoodsAttachment(zongpai_no=zongpai_no, determination="has")
|
||
db.add(att)
|
||
elif att.determination in ("undetermined", "none"):
|
||
att.determination = "has"
|
||
att.updated_at = datetime.now()
|
||
db.flush()
|
||
|
||
item = (
|
||
db.query(FinishedGoodsAttachmentItem)
|
||
.filter(
|
||
FinishedGoodsAttachmentItem.zongpai_no == zongpai_no,
|
||
FinishedGoodsAttachmentItem.category_id == category_id,
|
||
)
|
||
.first()
|
||
)
|
||
if item is None:
|
||
item = FinishedGoodsAttachmentItem(
|
||
zongpai_no=zongpai_no,
|
||
category_id=category_id,
|
||
expected_qty=erp_qty,
|
||
)
|
||
db.add(item)
|
||
db.flush()
|
||
return item
|
||
|
||
|
||
def save_attachment_box(db: Session, req) -> dict:
|
||
zongpai_no = _validate_zongpai(req.zongpai_no)
|
||
|
||
erp = query_erp_info(db, zongpai_no) # raises ZongpaiNotFoundError if missing
|
||
paichan_no = erp["paichan_no"]
|
||
erp_qty = erp["quantity"]
|
||
|
||
item = _ensure_attachment_item(db, zongpai_no, req.category_id, erp_qty)
|
||
|
||
box = (
|
||
db.query(FinishedGoodsBox)
|
||
.filter(
|
||
FinishedGoodsBox.paichan_no == paichan_no,
|
||
FinishedGoodsBox.box_no == req.box_no,
|
||
)
|
||
.first()
|
||
)
|
||
if box is not None:
|
||
existing_in_box = (
|
||
db.query(FinishedGoodsBoxAttachmentItem)
|
||
.filter(
|
||
FinishedGoodsBoxAttachmentItem.box_id == box.id,
|
||
FinishedGoodsBoxAttachmentItem.zongpai_no == zongpai_no,
|
||
FinishedGoodsBoxAttachmentItem.category_id == req.category_id,
|
||
)
|
||
.first()
|
||
)
|
||
if existing_in_box is not None:
|
||
raise DuplicateAttachmentBoxItemError(paichan_no, req.box_no)
|
||
else:
|
||
box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
|
||
db.add(box)
|
||
db.flush()
|
||
|
||
# Cap check against expected_qty (mirrors product _ensure_quantity_within_erp_total)
|
||
already_boxed = _boxed_for_category(db, zongpai_no, req.category_id)
|
||
if already_boxed + req.quantity > item.expected_qty:
|
||
raise InvalidQuantityError()
|
||
|
||
record = FinishedGoodsBoxAttachmentItem(
|
||
box_id=box.id,
|
||
zongpai_no=zongpai_no,
|
||
category_id=req.category_id,
|
||
quantity=req.quantity,
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
db.refresh(record)
|
||
|
||
return {
|
||
"box_item_id": record.id,
|
||
"paichan_no": paichan_no,
|
||
"box_no": req.box_no,
|
||
"zongpai_no": zongpai_no,
|
||
"category_id": req.category_id,
|
||
"quantity": req.quantity,
|
||
"created_at": record.created_at.isoformat(),
|
||
}
|
||
|
||
|
||
def update_attachment_box(db: Session, item_id: int, req) -> dict:
|
||
record = (
|
||
db.query(FinishedGoodsBoxAttachmentItem)
|
||
.filter(FinishedGoodsBoxAttachmentItem.id == item_id)
|
||
.first()
|
||
)
|
||
if record is None:
|
||
raise AttachmentItemNotFoundError()
|
||
|
||
current_box = (
|
||
db.query(FinishedGoodsBox)
|
||
.filter(FinishedGoodsBox.id == record.box_id)
|
||
.first()
|
||
)
|
||
if current_box is None:
|
||
raise AttachmentItemNotFoundError()
|
||
paichan_no = current_box.paichan_no
|
||
|
||
target_box = (
|
||
db.query(FinishedGoodsBox)
|
||
.filter(
|
||
FinishedGoodsBox.paichan_no == paichan_no,
|
||
FinishedGoodsBox.box_no == req.box_no,
|
||
)
|
||
.first()
|
||
)
|
||
if target_box is None:
|
||
target_box = FinishedGoodsBox(paichan_no=paichan_no, box_no=req.box_no)
|
||
db.add(target_box)
|
||
db.flush()
|
||
else:
|
||
dup = (
|
||
db.query(FinishedGoodsBoxAttachmentItem)
|
||
.filter(
|
||
FinishedGoodsBoxAttachmentItem.box_id == target_box.id,
|
||
FinishedGoodsBoxAttachmentItem.zongpai_no == record.zongpai_no,
|
||
FinishedGoodsBoxAttachmentItem.category_id == record.category_id,
|
||
FinishedGoodsBoxAttachmentItem.id != record.id,
|
||
)
|
||
.first()
|
||
)
|
||
if dup is not None:
|
||
raise DuplicateAttachmentBoxItemError(paichan_no, req.box_no)
|
||
|
||
item = (
|
||
db.query(FinishedGoodsAttachmentItem)
|
||
.filter(
|
||
FinishedGoodsAttachmentItem.zongpai_no == record.zongpai_no,
|
||
FinishedGoodsAttachmentItem.category_id == record.category_id,
|
||
)
|
||
.first()
|
||
)
|
||
expected_qty = item.expected_qty if item is not None else 0
|
||
already_boxed = _boxed_for_category(
|
||
db, record.zongpai_no, record.category_id, exclude_item_id=record.id
|
||
)
|
||
if already_boxed + req.quantity > expected_qty:
|
||
raise InvalidQuantityError()
|
||
|
||
old_box_id = record.box_id
|
||
record.box_id = target_box.id
|
||
record.quantity = req.quantity
|
||
|
||
if old_box_id != target_box.id:
|
||
_cleanup_empty_box(db, old_box_id, exclude_item_id=record.id)
|
||
|
||
db.commit()
|
||
db.refresh(record)
|
||
return {
|
||
"box_item_id": record.id,
|
||
"paichan_no": paichan_no,
|
||
"box_no": req.box_no,
|
||
"zongpai_no": record.zongpai_no,
|
||
"category_id": record.category_id,
|
||
"quantity": int(record.quantity or 0),
|
||
"updated_at": record.created_at.isoformat(),
|
||
}
|
||
|
||
|
||
def delete_attachment_box(db: Session, item_id: int) -> dict:
|
||
record = (
|
||
db.query(FinishedGoodsBoxAttachmentItem)
|
||
.filter(FinishedGoodsBoxAttachmentItem.id == item_id)
|
||
.first()
|
||
)
|
||
if record is None:
|
||
raise AttachmentItemNotFoundError()
|
||
box_id = record.box_id
|
||
db.delete(record)
|
||
db.flush()
|
||
_cleanup_empty_box(db, box_id)
|
||
db.commit()
|
||
return {"box_item_id": item_id, "deleted": True}
|
||
|
||
|
||
def _cleanup_empty_box(db: Session, box_id: int, exclude_item_id: int | None = None) -> None:
|
||
"""If a box has no product items and no attachment items, delete it."""
|
||
from app.models.finished_goods import FinishedGoodsBoxItem
|
||
|
||
product_q = db.query(FinishedGoodsBoxItem.id).filter(
|
||
FinishedGoodsBoxItem.box_id == box_id
|
||
)
|
||
if product_q.first() is not None:
|
||
return
|
||
att_q = db.query(FinishedGoodsBoxAttachmentItem.id).filter(
|
||
FinishedGoodsBoxAttachmentItem.box_id == box_id
|
||
)
|
||
if exclude_item_id is not None:
|
||
att_q = att_q.filter(FinishedGoodsBoxAttachmentItem.id != exclude_item_id)
|
||
if att_q.first() is not None:
|
||
return
|
||
box = db.query(FinishedGoodsBox).filter(FinishedGoodsBox.id == box_id).first()
|
||
if box is not None:
|
||
db.delete(box)
|