mirror of
https://github.com/frappe/erpnext.git
synced 2026-09-09 23:39:28 +00:00
370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""Resolve physical numbers at input boundaries. Stock references always contain document IDs."""
|
|
|
|
import frappe
|
|
from frappe import _
|
|
from frappe.model.naming import make_autoname
|
|
from frappe.query_builder.functions import Coalesce, Count, Lower, NullIf
|
|
from frappe.utils import cstr, now
|
|
|
|
from erpnext.stock.serial_batch_number_lookup import SerialBatchNumberLookup
|
|
|
|
|
|
class SerialBatchIdentity:
|
|
def __init__(self, doctype):
|
|
self.doctype = doctype
|
|
self.item_field, self.number_field = {
|
|
"Serial No": ("item_code", "serial_no"),
|
|
"Batch": ("item", "batch_id"),
|
|
}[doctype]
|
|
|
|
def resolve(self, item_code, numbers, *, create=False, defaults=None, check_permissions=False):
|
|
"""Return IDs in input order. A physical number is never looked up as a document ID."""
|
|
if not isinstance(numbers, list | tuple) or any(not isinstance(number, str) for number in numbers):
|
|
frappe.throw(_("Physical numbers must be a list of strings"))
|
|
numbers = [number.strip() for number in numbers]
|
|
if not numbers:
|
|
return []
|
|
if not isinstance(item_code, str) or not item_code or any(not number for number in numbers):
|
|
frappe.throw(_("Item and physical number are required"))
|
|
|
|
if check_permissions:
|
|
permission = "select" if frappe.only_has_select_perm(self.doctype) else "read"
|
|
frappe.has_permission(self.doctype, permission, throw=True)
|
|
lookup = SerialBatchNumberLookup(self, item_code, numbers)
|
|
missing = lookup.missing
|
|
if missing:
|
|
if not create:
|
|
frappe.throw(
|
|
_("{0} {1} does not exist for Item {2}").format(self.doctype, missing[0], item_code)
|
|
)
|
|
if check_permissions:
|
|
frappe.has_permission(self.doctype, "create", throw=True)
|
|
lookup.assign(self.resolve_missing(item_code, missing, defaults))
|
|
names = [lookup.ids[number] for number in numbers]
|
|
if check_permissions:
|
|
allowed = frappe.get_list(
|
|
self.doctype, filters={"name": ("in", names)}, pluck="name", limit_page_length=0
|
|
)
|
|
if set(names) - set(allowed):
|
|
frappe.throw(
|
|
_("Not permitted to select these serial or batch records"), frappe.PermissionError
|
|
)
|
|
return names
|
|
|
|
def resolve_missing(self, item_code, numbers, defaults):
|
|
savepoint = "serial_batch_resolve_" + frappe.generate_hash(length=10)
|
|
frappe.db.savepoint(savepoint)
|
|
try:
|
|
ids = self.create_many(item_code, numbers, defaults)
|
|
except Exception as error:
|
|
frappe.db.rollback(save_point=savepoint)
|
|
frappe.db.release_savepoint(savepoint)
|
|
if not isinstance(error, frappe.DuplicateEntryError | frappe.UniqueValidationError):
|
|
raise
|
|
else:
|
|
frappe.db.release_savepoint(savepoint)
|
|
return ids
|
|
|
|
lookup = SerialBatchNumberLookup(self, item_code, numbers)
|
|
if lookup.missing:
|
|
lookup.assign(self.create_many(item_code, lookup.missing, defaults))
|
|
return lookup.ids
|
|
|
|
def get_query(self, numbers, item_code=None, *, fields=None, filters=None, ignore_permissions=True):
|
|
table = frappe.qb.DocType(self.doctype)
|
|
filters = {**(filters or {}), **({self.item_field: item_code} if item_code else {})}
|
|
return frappe.qb.get_query(
|
|
self.doctype,
|
|
fields=fields or ["name", self.number_field],
|
|
filters=filters,
|
|
ignore_permissions=ignore_permissions,
|
|
).where(
|
|
self.number_key(table[self.number_field]).isin([self.number_key(number) for number in numbers])
|
|
)
|
|
|
|
def number_key(self, value):
|
|
# Preserve MariaDB's collation. PostgreSQL needs an explicit case-insensitive comparison.
|
|
return Lower(value) if frappe.db.db_type == "postgres" else value
|
|
|
|
def exists(self, number, item_code=None, *, exclude=None):
|
|
filters = {"name": ("!=", exclude)} if exclude else None
|
|
rows = self.get_query([number], item_code, fields=["name"], filters=filters).limit(1).run()
|
|
return rows[0][0] if rows else None
|
|
|
|
def create_many(self, item_code, numbers, defaults=None):
|
|
if self.doctype == "Batch":
|
|
# New batches still need their expiry and valuation hooks.
|
|
return {number: self.create_batch(item_code, number, defaults) for number in numbers}
|
|
|
|
# Inactive serials can be prepared before their first receipt assigns a company.
|
|
item = frappe.get_cached_value(
|
|
"Item", item_code, ["item_name", "description", "warranty_period", "has_serial_no"], as_dict=True
|
|
)
|
|
if not item.has_serial_no:
|
|
frappe.throw(_("Item {0} does not have serial numbers enabled").format(item_code))
|
|
common = {
|
|
"item_code": item_code,
|
|
"item_name": item.item_name,
|
|
"description": item.description,
|
|
"warranty_period": item.warranty_period or 0,
|
|
"status": "Inactive",
|
|
"creation": now(),
|
|
"modified": now(),
|
|
"owner": frappe.session.user,
|
|
"modified_by": frappe.session.user,
|
|
**(defaults or {}),
|
|
}
|
|
ids = {number: make_autoname("hash", "Serial No") for number in numbers}
|
|
try:
|
|
frappe.db.bulk_insert(
|
|
"Serial No",
|
|
fields=["name", "serial_no", *common],
|
|
values=[(name, number, *common.values()) for number, name in ids.items()],
|
|
)
|
|
except Exception as error:
|
|
if frappe.db.is_unique_key_violation(error) or frappe.db.is_primary_key_violation(error):
|
|
raise frappe.DuplicateEntryError(
|
|
_("A serial number already exists for Item {0}. Refresh and try again.").format(item_code)
|
|
) from error
|
|
raise
|
|
return ids
|
|
|
|
def create_batch(self, item_code, number, defaults=None):
|
|
doc = frappe.new_doc("Batch")
|
|
doc.update(defaults or {})
|
|
doc.item = item_code
|
|
doc.batch_id = number
|
|
doc.flags.serial_batch_number_checked = True
|
|
doc.insert(ignore_permissions=True)
|
|
return doc.name
|
|
|
|
def labels(self, names):
|
|
if not names:
|
|
return {}
|
|
return dict(
|
|
frappe.get_all(
|
|
self.doctype,
|
|
filters={"name": ("in", list(set(names)))},
|
|
fields=["name", self.number_field],
|
|
as_list=True,
|
|
)
|
|
)
|
|
|
|
def validate(self, doc):
|
|
number = cstr(doc.get(self.number_field)).strip()
|
|
doc.set(self.number_field, number)
|
|
if (
|
|
number
|
|
and not (doc.is_new() and doc.flags.serial_batch_number_checked)
|
|
and self.exists(number, doc.get(self.item_field), exclude=doc.name)
|
|
):
|
|
frappe.throw(
|
|
_("{0} {1} already exists for Item {2}").format(
|
|
self.doctype, number, doc.get(self.item_field)
|
|
),
|
|
frappe.DuplicateEntryError,
|
|
)
|
|
|
|
def backfill_numbers(self):
|
|
table = frappe.qb.DocType(self.doctype)
|
|
frappe.qb.update(table).set(table[self.number_field], table.name).where(
|
|
table[self.number_field].isnull() | (table[self.number_field] == "")
|
|
).run()
|
|
|
|
def has_constraint(self):
|
|
index = (
|
|
("serial_no_number_item_ci" if self.doctype == "Serial No" else "batch_number_item_ci")
|
|
if frappe.db.db_type == "postgres"
|
|
else f"unique_{self.item_field}_{self.number_field}"
|
|
)
|
|
return bool(frappe.db.has_index(f"tab{self.doctype}", index))
|
|
|
|
def sync_constraint(self):
|
|
if self.has_constraint():
|
|
return
|
|
if self.doctype not in (frappe.flags.serial_batch_preflight or ()):
|
|
self.validate_existing_numbers()
|
|
self.backfill_numbers()
|
|
if frappe.db.db_type == "postgres":
|
|
# The leading number expression also indexes scans without an item filter.
|
|
# add_unique only accepts column names, so expression indexes need explicit DDL.
|
|
frappe.db.sql_ddl(
|
|
{
|
|
"Serial No": 'CREATE UNIQUE INDEX IF NOT EXISTS "serial_no_number_item_ci" '
|
|
'ON "tabSerial No" (lower("serial_no"), "item_code")',
|
|
"Batch": 'CREATE UNIQUE INDEX IF NOT EXISTS "batch_number_item_ci" '
|
|
'ON "tabBatch" (lower("batch_id"), "item")',
|
|
}[self.doctype]
|
|
)
|
|
else:
|
|
frappe.db.add_unique(self.doctype, [self.item_field, self.number_field])
|
|
|
|
def validate_existing_numbers(self):
|
|
table = frappe.qb.DocType(self.doctype)
|
|
# Use the future backfilled value without changing legacy records during the preflight.
|
|
number = (
|
|
Coalesce(NullIf(table[self.number_field], ""), table.name)
|
|
if frappe.db.has_column(self.doctype, self.number_field)
|
|
else table.name
|
|
)
|
|
key = self.number_key(number)
|
|
duplicates = (
|
|
frappe.qb.from_(table)
|
|
.select(table[self.item_field], key)
|
|
.groupby(table[self.item_field], key)
|
|
.having(Count(table.name) > 1)
|
|
).run()
|
|
if not duplicates:
|
|
return
|
|
|
|
conflicts = []
|
|
for item, value in duplicates:
|
|
names = (
|
|
frappe.qb.from_(table)
|
|
.select(table.name)
|
|
.where((table[self.item_field] == item) & (key == value))
|
|
.orderby(table.name)
|
|
).run(pluck=True)
|
|
conflicts.append(_("Item {0}, number {1}: {2}").format(item, value, ", ".join(names)))
|
|
frappe.throw(
|
|
_(
|
|
"Resolve duplicate {0} physical numbers before upgrading. Document IDs and stock references have not been changed."
|
|
).format(self.doctype)
|
|
+ "\n"
|
|
+ "\n".join(conflicts),
|
|
title=_("Duplicate Serial or Batch Numbers"),
|
|
)
|
|
|
|
|
|
@frappe.whitelist(methods=["POST"])
|
|
def resolve_serial_batch_numbers(
|
|
item_code: str,
|
|
serial_numbers: list | str | None = None,
|
|
batch_numbers: list | str | None = None,
|
|
create: bool = False,
|
|
):
|
|
"""Resolve physical input to Link values. Existing ID-based APIs keep their meaning."""
|
|
frappe.has_permission("Item", "read", doc=item_code, throw=True)
|
|
result = {}
|
|
for doctype, values, key in (
|
|
("Serial No", serial_numbers, "serial_nos"),
|
|
("Batch", batch_numbers, "batch_nos"),
|
|
):
|
|
values = frappe.parse_json(values) or []
|
|
result[key] = SerialBatchIdentity(doctype).resolve(
|
|
item_code, values, create=create, check_permissions=True
|
|
)
|
|
return result
|
|
|
|
|
|
@frappe.whitelist(methods=["GET", "POST"])
|
|
def get_serial_batch_labels(doctype: str, names: list | str):
|
|
if doctype not in ("Serial No", "Batch"):
|
|
frappe.throw(_("Only Serial No and Batch labels are supported"))
|
|
names = frappe.parse_json(names)
|
|
if not isinstance(names, list) or any(not isinstance(name, str) for name in names):
|
|
frappe.throw(_("Document IDs must be a list of strings"))
|
|
identity = SerialBatchIdentity(doctype)
|
|
return dict(
|
|
frappe.get_list(
|
|
doctype,
|
|
filters={"name": ("in", names)},
|
|
fields=["name", identity.number_field],
|
|
as_list=True,
|
|
limit_page_length=0,
|
|
)
|
|
)
|
|
|
|
|
|
@frappe.whitelist(methods=["POST"])
|
|
def resolve_transaction_serial_numbers(parent: dict | str, row: dict | str, numbers: list | str):
|
|
parent, row = frappe.parse_json(parent), frappe.parse_json(row)
|
|
if not isinstance(parent, dict) or not isinstance(parent.get("doctype"), str):
|
|
frappe.throw(_("Transaction DocType is required"))
|
|
if not frappe.db.exists("DocType", parent["doctype"]):
|
|
frappe.throw(_("Invalid transaction DocType"))
|
|
if not isinstance(row, dict) or not (row.get("item_code") or row.get("rm_item_code")):
|
|
frappe.throw(_("Item is required"))
|
|
frappe.has_permission(
|
|
parent["doctype"],
|
|
"write",
|
|
doc=parent.get("name") if not parent.get("__islocal") else None,
|
|
throw=True,
|
|
)
|
|
return resolve_serial_batch_numbers(
|
|
row.get("item_code") or row.get("rm_item_code"), serial_numbers=numbers
|
|
)["serial_nos"]
|
|
|
|
|
|
def add_number_labels(entries):
|
|
"""Attach display values without changing the references or their field names."""
|
|
for field, doctype, label in (
|
|
("serial_no", "Serial No", "serial_number"),
|
|
("batch_no", "Batch", "batch_number"),
|
|
):
|
|
labels = SerialBatchIdentity(doctype).labels([row.get(field) for row in entries if row.get(field)])
|
|
for row in entries:
|
|
row[label] = labels.get(row.get(field), row.get(field))
|
|
return entries
|
|
|
|
|
|
def resolve_number_entries(item_code, entries, *, create=False):
|
|
"""Only explicit physical-number fields are resolved. Link fields already contain IDs."""
|
|
for field, doctype, number_field in (
|
|
("batch_no", "Batch", "batch_number"),
|
|
("serial_no", "Serial No", "serial_number"),
|
|
):
|
|
rows = [row for row in entries if row.get(number_field) and not row.get(field)]
|
|
ids = (
|
|
resolve_serial_batch_numbers(
|
|
item_code,
|
|
**{
|
|
"serial_numbers" if doctype == "Serial No" else "batch_numbers": [
|
|
row[number_field] for row in rows
|
|
]
|
|
},
|
|
create=create,
|
|
)["serial_nos" if doctype == "Serial No" else "batch_nos"]
|
|
if rows
|
|
else []
|
|
)
|
|
for row, name in zip(rows, ids, strict=True):
|
|
row[field] = name
|
|
serials = [row["serial_no"] for row in entries if row.get("serial_no") and not row.get("batch_no")]
|
|
batches = (
|
|
dict(
|
|
frappe.get_all(
|
|
"Serial No",
|
|
filters={"name": ("in", serials), "item_code": item_code},
|
|
fields=["name", "batch_no"],
|
|
as_list=True,
|
|
)
|
|
)
|
|
if serials
|
|
else {}
|
|
)
|
|
for row in entries:
|
|
if not row.get("batch_no") and row.get("serial_no") in batches and batches[row["serial_no"]]:
|
|
row["batch_no"] = batches[row["serial_no"]]
|
|
return entries
|
|
|
|
|
|
def validate_item_merge(old, new):
|
|
for doctype in ("Serial No", "Batch"):
|
|
identity = SerialBatchIdentity(doctype)
|
|
source = frappe.qb.DocType(doctype).as_("source")
|
|
target = frappe.qb.DocType(doctype).as_("target")
|
|
conflict = (
|
|
frappe.qb.from_(source)
|
|
.join(target)
|
|
.on(
|
|
identity.number_key(source[identity.number_field])
|
|
== identity.number_key(target[identity.number_field])
|
|
)
|
|
.select(source[identity.number_field])
|
|
.where((source[identity.item_field] == old) & (target[identity.item_field] == new))
|
|
.limit(1)
|
|
).run()
|
|
if conflict:
|
|
frappe.throw(_("Cannot merge items with the same {0}: {1}").format(doctype, conflict[0][0]))
|