wangbeihong/inventory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加锁仓模式设置，增强物料合并和编辑功能

Browse Source
This commit is contained in:
wangbeihong
2026-03-12 14:09:44 +08:00
parent 10da4c2859
commit 8928da3929
1 changed files with 409 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

489
app.py
View File

@@ -51,6 +51,7 @@ AI_SETTINGS_DEFAULT = {
"lcsc_app_id": os.environ.get("LCSC_APP_ID", ""),
"lcsc_access_key": os.environ.get("LCSC_ACCESS_KEY", ""),
"lcsc_secret_key": os.environ.get("LCSC_SECRET_KEY", ""),
"lock_storage_mode": False,
}
@@ -185,6 +186,7 @@ def _get_ai_settings() -> dict:
settings["lcsc_app_id"] = (settings.get("lcsc_app_id") or "").strip()
settings["lcsc_access_key"] = (settings.get("lcsc_access_key") or "").strip()
settings["lcsc_secret_key"] = (settings.get("lcsc_secret_key") or "").strip()
settings["lock_storage_mode"] = bool(settings.get("lock_storage_mode", False))
return settings
@@ -458,6 +460,137 @@ def slot_range_label(box: Box) -> str:
return f"{start_code}-{end_code}"
def _find_enabled_part_no_conflict(part_no: str, exclude_component_id: int = None):
normalized = (part_no or "").strip()
if not normalized:
return None
query = Component.query.filter(
Component.part_no == normalized,
Component.is_enabled.is_(True),
)
if exclude_component_id is not None:
query = query.filter(Component.id != exclude_component_id)
return query.order_by(Component.box_id.asc(), Component.slot_index.asc()).first()
def _normalize_material_text(text: str) -> str:
raw = (text or "").upper().strip()
if not raw:
return ""
raw = raw.replace("（", "(").replace("）", ")")
raw = re.sub(r"[\s\-_/,|;:，。；：]+", "", raw)
raw = re.sub(r"[()\[\]{}]", "", raw)
return raw
def _material_identity_key(name: str, specification: str) -> str:
name_key = _normalize_material_text(name)
spec_key = _normalize_material_text(specification)
if not name_key and not spec_key:
return ""
return f"{name_key}|{spec_key}"
def _find_enabled_material_conflict(
name: str,
specification: str,
exclude_component_id: int = None,
exclude_part_no: str = "",
):
target_key = _material_identity_key(name, specification)
if not target_key:
return None
query = Component.query.filter(Component.is_enabled.is_(True))
if exclude_component_id is not None:
query = query.filter(Component.id != exclude_component_id)
normalized_exclude_part_no = (exclude_part_no or "").strip()
for candidate in query.order_by(Component.box_id.asc(), Component.slot_index.asc()).all():
if normalized_exclude_part_no and (candidate.part_no or "").strip() == normalized_exclude_part_no:
continue
if _material_identity_key(candidate.name, candidate.specification) == target_key:
return candidate
return None
def _is_truthy_form_value(value: str) -> bool:
return (value or "").strip().lower() in {"1", "true", "yes", "on"}
def _append_merge_part_no_note(note: str, part_no: str) -> str:
normalized = (part_no or "").strip()
if not normalized:
return note or ""
line = f"合并料号: {normalized}"
current = note or ""
if line in current:
return current
return f"{current}\n{line}".strip()
def _is_slot_part_replacement(component: Component, new_part_no: str) -> bool:
if component is None:
return False
old_part_no = (component.part_no or "").strip()
target_part_no = (new_part_no or "").strip()
return bool(old_part_no and target_part_no and old_part_no != target_part_no)
def _merge_into_existing_component(
target: Component,
incoming_part_no: str,
incoming_name: str,
incoming_specification: str,
incoming_note: str,
incoming_quantity: int,
source_component: Component = None,
source_box: Box = None,
) -> None:
old_target_enabled_qty = int(target.quantity or 0) if target.is_enabled else 0
target.name = incoming_name or target.name
if incoming_specification:
target.specification = incoming_specification
if incoming_note:
target.note = incoming_note
if incoming_part_no and incoming_part_no != target.part_no:
target.note = _append_merge_part_no_note(target.note, incoming_part_no)
target.quantity = int(target.quantity or 0) + int(incoming_quantity or 0)
target.is_enabled = True
target_delta = int(target.quantity or 0) - old_target_enabled_qty
if target_delta:
log_inventory_event(
event_type="component_merge_confirmed",
delta=target_delta,
box=target.box,
component=target,
part_no=target.part_no,
)
if source_component is not None and source_component.id != target.id:
if source_component.is_enabled and source_component.quantity:
log_inventory_event(
event_type="component_merge_cleanup",
delta=-int(source_component.quantity),
box=source_box,
component=source_component,
part_no=source_component.part_no,
)
db.session.delete(source_component)
def _format_component_position(component: Component) -> str:
target_box = Box.query.get(component.box_id)
if not target_box:
return f"盒子#{component.box_id} 位置#{component.slot_index}"
return f"{target_box.name} {slot_code_for_box(target_box, component.slot_index)}"
def compose_box_name(base_name: str, prefix: str, start_number: int, slot_capacity: int) -> str:
base = (base_name or "").strip()
if not base:
@@ -1556,10 +1689,8 @@ def quick_inbound(box_id: int):
components = Component.query.filter_by(box_id=box.id).all()
occupied_slots = {c.slot_index for c in components}
by_part_no = {c.part_no: c for c in components if c.part_no}
added_count = 0
merged_count = 0
skipped_lines = []
changed = False
@@ -1581,28 +1712,24 @@ def quick_inbound(box_id: int):
skipped_lines.append(f"第 {line_no} 行")
continue
existing = by_part_no.get(part_no)
if existing:
old_enabled_qty = existing.quantity if existing.is_enabled else 0
existing.quantity += quantity
existing.name = name or existing.name
if specification:
existing.specification = specification
if note:
existing.note = note
existing.is_enabled = True
new_enabled_qty = existing.quantity
delta = new_enabled_qty - old_enabled_qty
if delta:
log_inventory_event(
event_type="quick_inbound_merge",
delta=delta,
box=box,
component=existing,
part_no=part_no,
)
merged_count += 1
changed = True
part_no_conflict = _find_enabled_part_no_conflict(part_no)
if part_no_conflict:
skipped_lines.append(
f"第 {line_no} 行({part_no} 已在 {_format_component_position(part_no_conflict)}，需人工确认后再合并)"
)
continue
material_conflict = _find_enabled_material_conflict(
name=name,
specification=specification,
exclude_part_no=part_no,
)
if material_conflict:
skipped_lines.append(
"第 "
f"{line_no} 行(检测到同参数物料: {material_conflict.part_no} 已在 "
f"{_format_component_position(material_conflict)}，需人工确认后再合并)"
)
continue
slot_index = _next_empty_slot_index(box, occupied_slots)
@@ -1630,20 +1757,19 @@ def quick_inbound(box_id: int):
part_no=part_no,
)
occupied_slots.add(slot_index)
by_part_no[part_no] = item
added_count += 1
changed = True
if changed:
db.session.commit()
if added_count == 0 and merged_count == 0:
if added_count == 0:
return render_box_page(
box,
error="快速入库失败: 没有可导入的数据，请检查格式",
)
message = f"快速入库完成: 新增 {added_count} 条，合并 {merged_count} 条"
message = f"快速入库完成: 新增 {added_count} 条"
if skipped_lines:
message += "；跳过: " + ", ".join(skipped_lines)
return render_box_page(box, notice=message)
@@ -1668,28 +1794,30 @@ def add_bag_item(box_id: int):
except ValueError:
return render_box_page(box, error="袋装新增失败: 数量必须是大于等于 0 的整数")
existing = Component.query.filter_by(box_id=box.id, part_no=part_no).first()
if existing:
old_enabled_qty = existing.quantity if existing.is_enabled else 0
existing.name = name or existing.name
existing.quantity += quantity
existing.specification = specification or existing.specification
existing.note = note or existing.note
existing.is_enabled = True
part_no_conflict = _find_enabled_part_no_conflict(part_no)
if part_no_conflict:
return render_box_page(
box,
error=(
f"袋装新增失败: 料号 {part_no} 已存在于 "
f"{_format_component_position(part_no_conflict)}，需人工确认后再合并"
),
)
new_enabled_qty = existing.quantity
delta = int(new_enabled_qty - old_enabled_qty)
if delta:
log_inventory_event(
event_type="bag_merge",
delta=delta,
box=box,
component=existing,
part_no=part_no,
)
db.session.commit()
return render_box_page(box, notice="同料号已存在，已合并到原袋位")
material_conflict = _find_enabled_material_conflict(
name=name,
specification=specification,
exclude_part_no=part_no,
)
if material_conflict:
return render_box_page(
box,
error=(
"袋装新增失败: 检测到同参数物料 "
f"{material_conflict.part_no} 已在 "
f"{_format_component_position(material_conflict)}，需人工确认后再合并"
),
)
next_slot = (
db.session.query(db.func.max(Component.slot_index))
@@ -1736,7 +1864,7 @@ def add_bag_items_batch(box_id: int):
if not lines:
return render_box_page(box, error="批量新增失败: 请至少输入一行")
invalid_lines = []
skipped_lines = []
added_count = 0
next_slot = (
db.session.query(db.func.max(Component.slot_index))
@@ -1744,11 +1872,6 @@ def add_bag_items_batch(box_id: int):
.scalar()
or 0
) + 1
existing_by_part_no = {
c.part_no: c for c in Component.query.filter_by(box_id=box.id).all() if c.part_no
}
merged_count = 0
for line_no, line in enumerate(lines, start=1):
parts = [p.strip() for p in re.split(r"[,\t]", line)]
while len(parts) < 5:
@@ -1761,34 +1884,33 @@ def add_bag_items_batch(box_id: int):
note = parts[4]
if not part_no or not name:
invalid_lines.append(f"第 {line_no} 行")
skipped_lines.append(f"第 {line_no} 行(料号或名称为空)")
continue
try:
quantity = _parse_non_negative_int(quantity_raw, 0)
except ValueError:
invalid_lines.append(f"第 {line_no} 行")
skipped_lines.append(f"第 {line_no} 行(数量格式错误)")
continue
existing = existing_by_part_no.get(part_no)
if existing:
old_enabled_qty = existing.quantity if existing.is_enabled else 0
existing.name = name or existing.name
existing.quantity += quantity
existing.specification = specification or existing.specification
existing.note = note or existing.note
existing.is_enabled = True
part_no_conflict = _find_enabled_part_no_conflict(part_no)
if part_no_conflict:
skipped_lines.append(
f"第 {line_no} 行({part_no} 已在 {_format_component_position(part_no_conflict)}，需人工确认后再合并)"
)
continue
delta = int(existing.quantity - old_enabled_qty)
if delta:
log_inventory_event(
event_type="bag_batch_merge",
delta=delta,
box=box,
component=existing,
part_no=part_no,
)
merged_count += 1
material_conflict = _find_enabled_material_conflict(
name=name,
specification=specification,
exclude_part_no=part_no,
)
if material_conflict:
skipped_lines.append(
"第 "
f"{line_no} 行(检测到同参数物料: {material_conflict.part_no} 已在 "
f"{_format_component_position(material_conflict)}，需人工确认后再合并)"
)
continue
new_component = Component(
@@ -1802,7 +1924,6 @@ def add_bag_items_batch(box_id: int):
is_enabled=True,
)
db.session.add(new_component)
existing_by_part_no[part_no] = new_component
if quantity:
log_inventory_event(
event_type="bag_batch_add",
@@ -1818,19 +1939,19 @@ def add_bag_items_batch(box_id: int):
box.slot_capacity = max(box.slot_capacity, next_slot - 1)
db.session.commit()
if invalid_lines and added_count == 0 and merged_count == 0:
if skipped_lines and added_count == 0:
return render_box_page(
box,
error="批量新增失败: 数据格式不正确（请检查: " + ", ".join(invalid_lines) + "）",
error="批量新增失败: 没有可导入的数据（请检查: " + ", ".join(skipped_lines) + "）",
)
if invalid_lines:
if skipped_lines:
return render_box_page(
box,
notice=f"已新增 {added_count} 条，合并 {merged_count} 条，以下行被跳过: " + ", ".join(invalid_lines),
notice=f"已新增 {added_count} 条，以下行被跳过: " + ", ".join(skipped_lines),
)
return render_box_page(box, notice=f"批量新增成功：新增 {added_count} 条，合并 {merged_count} 条")
return render_box_page(box, notice=f"批量新增成功：新增 {added_count} 条")
@app.route("/edit/<int:box_id>/<int:slot>", methods=["GET", "POST"])
@@ -1843,13 +1964,43 @@ def edit_component(box_id: int, slot: int):
notice = request.args.get("notice", "").strip()
error = request.args.get("error", "").strip()
component = Component.query.filter_by(box_id=box.id, slot_index=slot).first()
settings = _get_ai_settings()
lock_storage_mode = bool(settings.get("lock_storage_mode", False))
if request.method == "POST":
action = request.form.get("action", "save")
search_query_post = request.form.get("q", "").strip()
search_query_effective = search_query_post or search_query
delete_confirm_slot = request.form.get("delete_confirm_slot", "").strip().upper()
expected_slot_code = slot_code_for_box(box, slot).strip().upper()
if action == "delete":
if lock_storage_mode:
return render_template(
"edit.html",
box=box,
slot=slot,
slot_code=slot_code_for_box(box, slot),
component=component,
error="锁仓模式已开启，禁止删除位置绑定。",
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
if delete_confirm_slot != expected_slot_code:
return render_template(
"edit.html",
box=box,
slot=slot,
slot_code=slot_code_for_box(box, slot),
component=component,
error=f"删除确认失败: 请输入当前位置编号 {expected_slot_code}",
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
if component:
if component.is_enabled and component.quantity:
log_inventory_event(
@@ -1897,6 +2048,8 @@ def edit_component(box_id: int, slot: int):
specification = request.form.get("specification", "").strip()
quantity_raw = request.form.get("quantity", "0").strip()
note = request.form.get("note", "").strip()
confirm_merge = _is_truthy_form_value(request.form.get("confirm_merge", ""))
confirm_position_change = _is_truthy_form_value(request.form.get("confirm_position_change", ""))
if not part_no or not name:
error = "料号和名称不能为空"
@@ -1909,6 +2062,7 @@ def edit_component(box_id: int, slot: int):
error=error,
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
try:
@@ -1924,8 +2078,97 @@ def edit_component(box_id: int, slot: int):
error=error,
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
if lock_storage_mode and _is_slot_part_replacement(component, part_no):
error = "锁仓模式已开启，禁止替换当前位置绑定料号。"
return render_template(
"edit.html",
box=box,
slot=slot,
slot_code=slot_code_for_box(box, slot),
component=component,
error=error,
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
if _is_slot_part_replacement(component, part_no) and not confirm_position_change:
error = (
"当前位已绑定到固定料号，检测到替换操作。"
"如需变更位置绑定，请勾选“我确认替换当前位物料”后再保存"
)
return render_template(
"edit.html",
box=box,
slot=slot,
slot_code=slot_code_for_box(box, slot),
component=component,
error=error,
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
current_component_id = component.id if component is not None else None
part_no_conflict = _find_enabled_part_no_conflict(part_no, exclude_component_id=current_component_id)
material_conflict = _find_enabled_material_conflict(
name=name,
specification=specification,
exclude_component_id=current_component_id,
exclude_part_no=part_no,
)
conflict = part_no_conflict or material_conflict
if conflict and not confirm_merge:
conflict_reason = "同料号" if part_no_conflict else "同参数"
error = (
f"检测到{conflict_reason}物料已存在于 {_format_component_position(conflict)}；"
"请勾选“人工确认后合并”再保存"
)
return render_template(
"edit.html",
box=box,
slot=slot,
slot_code=slot_code_for_box(box, slot),
component=component,
error=error,
notice=notice,
search_query=search_query_post,
lock_storage_mode=lock_storage_mode,
)
if conflict and confirm_merge:
_merge_into_existing_component(
target=conflict,
incoming_part_no=part_no,
incoming_name=name,
incoming_specification=specification,
incoming_note=note,
incoming_quantity=quantity,
source_component=component,
source_box=box,
)
db.session.commit()
target_box = Box.query.get(conflict.box_id)
target_slot = conflict.slot_index
notice_text = (
f"已人工确认合并到 {_format_component_position(conflict)}，"
f"累计数量 {conflict.quantity}"
)
if target_box:
return redirect(
url_for(
"edit_component",
box_id=target_box.id,
slot=target_slot,
notice=notice_text,
)
)
return redirect(url_for("view_box", box_id=box.id, notice=notice_text))
old_enabled_qty = 0
if component is not None and component.is_enabled:
old_enabled_qty = int(component.quantity)
@@ -1967,6 +2210,7 @@ def edit_component(box_id: int, slot: int):
error=error,
notice=notice,
search_query=search_query,
lock_storage_mode=lock_storage_mode,
)
@@ -1993,6 +2237,88 @@ def lcsc_import_to_edit_slot(box_id: int, slot: int):
return redirect(url_for("edit_component", box_id=box.id, slot=slot, error="立创导入失败: 商品信息缺少料号或名称"))
component = Component.query.filter_by(box_id=box.id, slot_index=slot).first()
confirm_merge = _is_truthy_form_value(request.form.get("confirm_merge", ""))
confirm_position_change = _is_truthy_form_value(request.form.get("confirm_position_change", ""))
if bool(settings.get("lock_storage_mode", False)) and _is_slot_part_replacement(component, mapped["part_no"]):
return redirect(
url_for(
"edit_component",
box_id=box.id,
slot=slot,
error="立创导入失败: 锁仓模式已开启，禁止替换当前位置绑定料号。",
)
)
if _is_slot_part_replacement(component, mapped["part_no"]) and not confirm_position_change:
return redirect(
url_for(
"edit_component",
box_id=box.id,
slot=slot,
error=(
"立创导入失败: 当前位已绑定固定料号。"
"如需替换，请勾选“我确认替换当前位物料”后再导入"
),
)
)
current_component_id = component.id if component is not None else None
part_no_conflict = _find_enabled_part_no_conflict(
mapped["part_no"],
exclude_component_id=current_component_id,
)
material_conflict = _find_enabled_material_conflict(
name=mapped["name"],
specification=mapped["specification"],
exclude_component_id=current_component_id,
exclude_part_no=mapped["part_no"],
)
conflict = part_no_conflict or material_conflict
if conflict and not confirm_merge:
conflict_reason = "同料号" if part_no_conflict else "同参数"
return redirect(
url_for(
"edit_component",
box_id=box.id,
slot=slot,
error=(
f"立创导入失败: 检测到{conflict_reason}物料已存在于 "
f"{_format_component_position(conflict)}，请勾选人工确认后合并"
),
)
)
if conflict and confirm_merge:
_merge_into_existing_component(
target=conflict,
incoming_part_no=mapped["part_no"],
incoming_name=mapped["name"],
incoming_specification=mapped["specification"],
incoming_note=mapped["note"],
incoming_quantity=quantity,
source_component=component,
source_box=box,
)
db.session.commit()
target_box = Box.query.get(conflict.box_id)
target_slot = conflict.slot_index
notice_text = (
f"已人工确认合并到 {_format_component_position(conflict)}，"
f"累计数量 {conflict.quantity}"
)
if target_box:
return redirect(
url_for(
"edit_component",
box_id=target_box.id,
slot=target_slot,
notice=notice_text,
)
)
return redirect(url_for("view_box", box_id=box.id, notice=notice_text))
old_enabled_qty = int(component.quantity or 0) if component and component.is_enabled else 0
if component is None:
@@ -2225,6 +2551,7 @@ def ai_settings_page():
lcsc_app_id = request.form.get("lcsc_app_id", "").strip()
lcsc_access_key = request.form.get("lcsc_access_key", "").strip()
lcsc_secret_key = request.form.get("lcsc_secret_key", "").strip()
lock_storage_mode = _is_truthy_form_value(request.form.get("lock_storage_mode", ""))
try:
timeout = int((request.form.get("timeout", "30") or "30").strip())
@@ -2282,6 +2609,7 @@ def ai_settings_page():
"lcsc_app_id": lcsc_app_id,
"lcsc_access_key": lcsc_access_key,
"lcsc_secret_key": lcsc_secret_key,
"lock_storage_mode": lock_storage_mode,
}
_save_ai_settings(settings)
return redirect(url_for("ai_settings_page", notice="AI参数已保存"))
@@ -2300,6 +2628,7 @@ def ai_settings_page():
"lcsc_app_id": lcsc_app_id,
"lcsc_access_key": lcsc_access_key,
"lcsc_secret_key": lcsc_secret_key,
"lock_storage_mode": lock_storage_mode,
}
)