133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
import re
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.attachment import (
|
|
AttachmentCategory,
|
|
FinishedGoodsAttachment,
|
|
FinishedGoodsAttachmentItem,
|
|
FinishedGoodsAttachmentLocation,
|
|
FinishedGoodsBoxAttachmentItem,
|
|
)
|
|
from app.services.box_service import InvalidZongpaiError
|
|
|
|
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,
|
|
}
|