mirror of
https://github.com/frappe/erpnext.git
synced 2026-08-16 07:58:38 +00:00
c0cfe5f363
Item Attribute abbreviations only got baked into a variant's item_code and item_name at creation time (make_variant_item_code returns early once item_code is set). Renaming an abbreviation afterwards left every existing variant stuck with the stale code, silently out of sync with its own attribute. Detect abbreviation renames on Item Attribute save, find every variant using the affected value, and rebuild+rename its item_code via frappe.rename_doc so linked records follow along. item_name is rebuilt in lockstep from the template's item_name, even if it had since been customized, since both fields are meant to be derived from the same abbreviation.
554 lines
18 KiB
Python
554 lines
18 KiB
Python
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
|
# License: GNU General Public License v3. See license.txt
|
|
|
|
|
|
import copy
|
|
import json
|
|
|
|
import frappe
|
|
from frappe import _
|
|
from frappe.query_builder import Case
|
|
from frappe.utils import cstr, flt
|
|
|
|
from erpnext.utilities.product import get_item_codes_by_attributes
|
|
|
|
|
|
class ItemVariantExistsError(frappe.ValidationError):
|
|
pass
|
|
|
|
|
|
class InvalidItemAttributeValueError(frappe.ValidationError):
|
|
pass
|
|
|
|
|
|
class ItemTemplateCannotHaveStock(frappe.ValidationError):
|
|
pass
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_variant(
|
|
template: str,
|
|
args: dict | str | None = None,
|
|
variant: str | None = None,
|
|
manufacturer: str | None = None,
|
|
manufacturer_part_no: str | None = None,
|
|
):
|
|
"""
|
|
Validates Attributes and their Values, then looks for an exactly
|
|
matching Item Variant
|
|
|
|
:param item: Template Item
|
|
:param args: A dictionary with "Attribute" as key and "Attribute Value" as value
|
|
"""
|
|
item_template = frappe.get_doc("Item", template)
|
|
|
|
if item_template.variant_based_on == "Manufacturer" and manufacturer:
|
|
return make_variant_based_on_manufacturer(item_template, manufacturer, manufacturer_part_no)
|
|
|
|
args = frappe.parse_json(args)
|
|
|
|
attribute_args = {k: v for k, v in args.items() if k != "use_template_image"}
|
|
if not attribute_args:
|
|
frappe.throw(_("Please specify at least one attribute in the Attributes table"))
|
|
|
|
return find_variant(template, args, variant)
|
|
|
|
|
|
def make_variant_based_on_manufacturer(template, manufacturer, manufacturer_part_no):
|
|
"""Make and return a new variant based on manufacturer and
|
|
manufacturer part no"""
|
|
from frappe.model.naming import append_number_if_name_exists
|
|
|
|
variant = frappe.new_doc("Item")
|
|
|
|
copy_attributes_to_variant(template, variant)
|
|
|
|
variant_name = f"{template.name} - {manufacturer}"
|
|
if manufacturer_part_no:
|
|
variant_name += f" - {manufacturer_part_no}"
|
|
|
|
variant.item_code = append_number_if_name_exists("Item", variant_name)
|
|
variant.flags.ignore_mandatory = True
|
|
variant.save()
|
|
|
|
if not frappe.db.exists("Item Manufacturer", {"item_code": variant.name, "manufacturer": manufacturer}):
|
|
manufacturer_doc = frappe.new_doc("Item Manufacturer")
|
|
manufacturer_doc.update(
|
|
{
|
|
"item_code": variant.name,
|
|
"manufacturer": manufacturer,
|
|
"manufacturer_part_no": manufacturer_part_no,
|
|
}
|
|
)
|
|
|
|
manufacturer_doc.flags.ignore_mandatory = True
|
|
manufacturer_doc.save(ignore_permissions=True)
|
|
|
|
return variant
|
|
|
|
|
|
def validate_item_variant_attributes(item, args=None):
|
|
if isinstance(item, str):
|
|
item = frappe.get_doc("Item", item)
|
|
|
|
if not args:
|
|
args = {d.attribute.lower(): d.attribute_value for d in item.attributes}
|
|
|
|
attribute_values, numeric_values = get_attribute_values(item)
|
|
|
|
for attribute, value in args.items():
|
|
if not value:
|
|
continue
|
|
|
|
if attribute.lower() in numeric_values:
|
|
numeric_attribute = numeric_values[attribute.lower()]
|
|
validate_is_incremental(numeric_attribute, attribute, value, item.name)
|
|
|
|
else:
|
|
attributes_list = attribute_values.get(attribute.lower(), [])
|
|
validate_item_attribute_value(attributes_list, attribute, value, item.name, from_variant=True)
|
|
|
|
|
|
def validate_is_incremental(numeric_attribute, attribute, value, item):
|
|
from_range = numeric_attribute.from_range
|
|
to_range = numeric_attribute.to_range
|
|
increment = numeric_attribute.increment
|
|
|
|
if increment == 0:
|
|
# defensive validation to prevent ZeroDivisionError
|
|
frappe.throw(_("Increment for Attribute {0} cannot be 0").format(attribute))
|
|
|
|
is_in_range = from_range <= flt(value) <= to_range
|
|
precision = max(len(cstr(v).split(".")[-1].rstrip("0")) for v in (value, increment))
|
|
# avoid precision error by rounding the remainder
|
|
remainder = flt((flt(value) - from_range) % increment, precision)
|
|
|
|
is_incremental = remainder == 0 or remainder == increment
|
|
|
|
if not (is_in_range and is_incremental):
|
|
frappe.throw(
|
|
_(
|
|
"Value for Attribute {0} must be within the range of {1} to {2} in the increments of {3} for Item {4}"
|
|
).format(attribute, from_range, to_range, increment, item),
|
|
InvalidItemAttributeValueError,
|
|
title=_("Invalid Attribute"),
|
|
)
|
|
|
|
|
|
def get_attribute_value_renames(item_attribute):
|
|
"""Return old to new attribute value mappings for renamed Item Attribute Value rows."""
|
|
if item_attribute.numeric_values:
|
|
return {}
|
|
|
|
db_value = item_attribute.get_doc_before_save()
|
|
if not db_value:
|
|
return {}
|
|
|
|
old_values = {d.name: d.attribute_value for d in db_value.item_attribute_values}
|
|
renames = {}
|
|
|
|
for row in item_attribute.item_attribute_values:
|
|
if row.name in old_values and old_values[row.name] != row.attribute_value:
|
|
renames[old_values[row.name]] = row.attribute_value
|
|
|
|
return renames
|
|
|
|
|
|
def update_variant_attribute_values(item_attribute):
|
|
"""Propagate renamed Item Attribute Values to Item Variant Attribute on variant items."""
|
|
value_map = get_attribute_value_renames(item_attribute)
|
|
if not value_map:
|
|
return
|
|
|
|
item_variant_table = frappe.qb.DocType("Item Variant Attribute")
|
|
item_table = frappe.qb.DocType("Item")
|
|
attribute_value = item_variant_table.attribute_value
|
|
attribute_value_case = Case()
|
|
|
|
for old_value, new_value in value_map.items():
|
|
attribute_value_case = attribute_value_case.when(attribute_value == old_value, new_value)
|
|
|
|
# Postgres has no UPDATE ... JOIN; restrict to variant items via a subquery on the parent instead.
|
|
variant_items = (
|
|
frappe.qb.from_(item_table)
|
|
.select(item_table.name)
|
|
.where(item_table.variant_of.isnotnull())
|
|
.where(item_table.variant_of != "")
|
|
)
|
|
(
|
|
frappe.qb.update(item_variant_table)
|
|
.set(attribute_value, attribute_value_case.else_(attribute_value))
|
|
.where(item_variant_table.parent.isin(variant_items))
|
|
.where(item_variant_table.attribute == item_attribute.name)
|
|
.where(attribute_value.isin(list(value_map)))
|
|
).run()
|
|
|
|
frappe.flags.attribute_values = None
|
|
|
|
|
|
def get_attribute_abbr_renames(item_attribute):
|
|
"""Return the set of (current) attribute values whose abbreviation was renamed."""
|
|
if item_attribute.numeric_values:
|
|
return set()
|
|
|
|
db_value = item_attribute.get_doc_before_save()
|
|
if not db_value:
|
|
return set()
|
|
|
|
old_abbrs = {d.name: d.abbr for d in db_value.item_attribute_values}
|
|
changed_values = set()
|
|
|
|
for row in item_attribute.item_attribute_values:
|
|
if row.name in old_abbrs and old_abbrs[row.name] != row.abbr:
|
|
changed_values.add(row.attribute_value)
|
|
|
|
return changed_values
|
|
|
|
|
|
def update_variant_item_codes_for_abbr_renames(item_attribute):
|
|
"""Rebuild item_code/item_name of variant Items affected by a renamed Item Attribute abbreviation."""
|
|
changed_values = get_attribute_abbr_renames(item_attribute)
|
|
if not changed_values:
|
|
return
|
|
|
|
item_variant_table = frappe.qb.DocType("Item Variant Attribute")
|
|
variant_names = (
|
|
frappe.qb.from_(item_variant_table)
|
|
.select(item_variant_table.parent)
|
|
.where(item_variant_table.attribute == item_attribute.name)
|
|
.where(item_variant_table.attribute_value.isin(list(changed_values)))
|
|
.distinct()
|
|
.run(pluck=True)
|
|
)
|
|
|
|
for variant_name in variant_names:
|
|
rename_variant_item_code(variant_name)
|
|
|
|
|
|
def rename_variant_item_code(variant_name):
|
|
"""Recompute a variant's item_code/item_name from its template and current attribute abbreviations,
|
|
renaming the Item if it has changed."""
|
|
variant = frappe.get_doc("Item", variant_name)
|
|
if not variant.variant_of:
|
|
return
|
|
|
|
template = frappe.get_cached_doc("Item", variant.variant_of)
|
|
|
|
new_code = frappe._dict({"item_code": None, "item_name": None, "attributes": variant.attributes})
|
|
make_variant_item_code(template.item_code, template.item_name, new_code)
|
|
|
|
if not new_code.item_code or new_code.item_code == variant.item_code:
|
|
return
|
|
|
|
frappe.rename_doc("Item", variant.item_code, new_code.item_code)
|
|
|
|
# Keep item_name in lockstep with item_code: both are derived from the same abbreviation, so
|
|
# item_name is always rebuilt here too, even if it had since been customized away from that pattern.
|
|
if new_code.item_name and new_code.item_name != variant.item_name:
|
|
frappe.db.set_value("Item", new_code.item_code, "item_name", new_code.item_name)
|
|
|
|
|
|
def validate_item_attribute_value(attributes_list, attribute, attribute_value, item, from_variant=True):
|
|
allow_rename_attribute_value = frappe.db.get_single_value(
|
|
"Item Variant Settings", "allow_rename_attribute_value"
|
|
)
|
|
if allow_rename_attribute_value:
|
|
pass
|
|
elif attribute_value not in attributes_list:
|
|
if from_variant:
|
|
frappe.throw(
|
|
_("{0} is not a valid Value for Attribute {1} of Item {2}.").format(
|
|
frappe.bold(attribute_value), frappe.bold(attribute), frappe.bold(item)
|
|
),
|
|
InvalidItemAttributeValueError,
|
|
title=_("Invalid Value"),
|
|
)
|
|
else:
|
|
msg = _("The value {0} is already assigned to an existing Item {1}.").format(
|
|
frappe.bold(attribute_value), frappe.bold(item)
|
|
)
|
|
msg += "<br>" + _(
|
|
"To still proceed with editing this Attribute Value, enable {0} in Item Variant Settings."
|
|
).format(frappe.bold(_("Allow Rename Attribute Value")))
|
|
|
|
frappe.throw(msg, InvalidItemAttributeValueError, title=_("Edit Not Allowed"))
|
|
|
|
|
|
def get_attribute_values(item):
|
|
if not frappe.flags.attribute_values:
|
|
attribute_values = {}
|
|
numeric_values = {}
|
|
for t in frappe.get_all("Item Attribute Value", fields=["parent", "attribute_value"]):
|
|
attribute_values.setdefault(t.parent.lower(), []).append(t.attribute_value)
|
|
|
|
for t in frappe.get_all(
|
|
"Item Variant Attribute",
|
|
fields=["attribute", "from_range", "to_range", "increment"],
|
|
filters={"numeric_values": 1, "parent": item.variant_of},
|
|
):
|
|
numeric_values[t.attribute.lower()] = t
|
|
|
|
frappe.flags.attribute_values = attribute_values
|
|
frappe.flags.numeric_values = numeric_values
|
|
|
|
return frappe.flags.attribute_values, frappe.flags.numeric_values
|
|
|
|
|
|
def find_variant(template, args, variant_item_code=None):
|
|
possible_variants = [i for i in get_item_codes_by_attributes(args, template) if i != variant_item_code]
|
|
|
|
for variant in possible_variants:
|
|
variant = frappe.get_doc("Item", variant)
|
|
|
|
if len(args.keys()) == len(variant.get("attributes")):
|
|
# has the same number of attributes and values
|
|
# assuming no duplication as per the validation in Item
|
|
match_count = 0
|
|
|
|
for attribute, value in args.items():
|
|
for row in variant.attributes:
|
|
if row.attribute == _(attribute) and row.attribute_value == cstr(value):
|
|
# this row matches
|
|
match_count += 1
|
|
break
|
|
|
|
if match_count == len(args.keys()):
|
|
return variant.name
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_variant(item: str, args: dict | str, use_template_image: bool = False):
|
|
use_template_image = frappe.parse_json(use_template_image)
|
|
args = frappe.parse_json(args)
|
|
|
|
template = frappe.get_doc("Item", item)
|
|
variant = frappe.new_doc("Item")
|
|
variant.variant_based_on = "Item Attribute"
|
|
variant_attributes = []
|
|
|
|
for d in template.attributes:
|
|
attribute_value = args.get(_(d.attribute)) or args.get(d.attribute)
|
|
if attribute_value:
|
|
variant_attributes.append({"attribute": d.attribute, "attribute_value": attribute_value})
|
|
|
|
variant.set("attributes", variant_attributes)
|
|
copy_attributes_to_variant(template, variant)
|
|
|
|
if use_template_image and template.image:
|
|
variant.image = template.image
|
|
|
|
make_variant_item_code(template.item_code, template.item_name, variant)
|
|
|
|
return variant
|
|
|
|
|
|
@frappe.whitelist()
|
|
def enqueue_multiple_variant_creation(item: str, args: dict | str, use_template_image: bool = False):
|
|
use_template_image = frappe.parse_json(use_template_image)
|
|
# There can be innumerable attribute combinations, enqueue
|
|
variants = frappe.parse_json(args)
|
|
variants = {key: values for key, values in variants.items() if values}
|
|
if not variants:
|
|
frappe.throw(_("Please select at least one attribute value"))
|
|
|
|
total_variants = 1
|
|
for key in variants:
|
|
total_variants *= len(variants[key])
|
|
if total_variants >= 600:
|
|
frappe.throw(_("Please do not create more than 500 items at a time"))
|
|
return
|
|
if total_variants < 10:
|
|
return create_multiple_variants(item, args, use_template_image)
|
|
else:
|
|
frappe.enqueue(
|
|
"erpnext.controllers.item_variant.create_multiple_variants",
|
|
item=item,
|
|
args=args,
|
|
use_template_image=use_template_image,
|
|
now=frappe.in_test,
|
|
)
|
|
return "queued"
|
|
|
|
|
|
def create_multiple_variants(item, args, use_template_image=False):
|
|
count = 0
|
|
args = frappe.parse_json(args)
|
|
args = {key: values for key, values in args.items() if values}
|
|
|
|
template_item = frappe.get_doc("Item", item)
|
|
args_set = generate_keyed_value_combinations(args)
|
|
|
|
for attribute_values in args_set:
|
|
if not get_variant(item, args=attribute_values):
|
|
variant = create_variant(item, attribute_values)
|
|
if use_template_image and template_item.image:
|
|
variant.image = template_item.image
|
|
variant.save()
|
|
count += 1
|
|
|
|
return count
|
|
|
|
|
|
def generate_keyed_value_combinations(args):
|
|
"""
|
|
From this:
|
|
|
|
args = {"attr1": ["a", "b", "c"], "attr2": ["1", "2"], "attr3": ["A"]}
|
|
|
|
To this:
|
|
|
|
[
|
|
{u'attr1': u'a', u'attr2': u'1', u'attr3': u'A'},
|
|
{u'attr1': u'b', u'attr2': u'1', u'attr3': u'A'},
|
|
{u'attr1': u'c', u'attr2': u'1', u'attr3': u'A'},
|
|
{u'attr1': u'a', u'attr2': u'2', u'attr3': u'A'},
|
|
{u'attr1': u'b', u'attr2': u'2', u'attr3': u'A'},
|
|
{u'attr1': u'c', u'attr2': u'2', u'attr3': u'A'}
|
|
]
|
|
|
|
"""
|
|
# Return empty list if empty
|
|
if not args:
|
|
return []
|
|
args = {key: values for key, values in args.items() if values}
|
|
if not args:
|
|
return []
|
|
|
|
# Turn `args` into a list of lists of key-value tuples:
|
|
# [
|
|
# [(u'attr2', u'1'), (u'attr2', u'2')],
|
|
# [(u'attr3', u'A')],
|
|
# [(u'attr1', u'a'), (u'attr1', u'b'), (u'attr1', u'c')]
|
|
# ]
|
|
key_value_lists = [[(key, val) for val in args[key]] for key in args.keys()]
|
|
|
|
# Store the first, but as objects
|
|
# [{u'attr2': u'1'}, {u'attr2': u'2'}]
|
|
results = key_value_lists.pop(0)
|
|
results = [{d[0]: d[1]} for d in results]
|
|
|
|
# Iterate the remaining
|
|
# Take the next list to fuse with existing results
|
|
for l in key_value_lists:
|
|
new_results = []
|
|
for res in results:
|
|
for key_val in l:
|
|
# create a new clone of object in result
|
|
obj = copy.deepcopy(res)
|
|
# to be used with every incoming new value
|
|
obj[key_val[0]] = key_val[1]
|
|
# and pushed into new_results
|
|
new_results.append(obj)
|
|
results = new_results
|
|
|
|
return results
|
|
|
|
|
|
def copy_attributes_to_variant(item, variant):
|
|
# copy non no-copy fields
|
|
|
|
exclude_fields = [
|
|
"naming_series",
|
|
"item_code",
|
|
"item_name",
|
|
"published_in_website",
|
|
"opening_stock",
|
|
"variant_of",
|
|
"valuation_rate",
|
|
]
|
|
|
|
if item.variant_based_on == "Manufacturer":
|
|
# don't copy manufacturer values if based on part no
|
|
exclude_fields += ["manufacturer", "manufacturer_part_no"]
|
|
|
|
allow_fields = [d.field_name for d in frappe.get_all("Variant Field", fields=["field_name"])]
|
|
if "variant_based_on" not in allow_fields:
|
|
allow_fields.append("variant_based_on")
|
|
for field in item.meta.fields:
|
|
# "Table" is part of `no_value_field` but we shouldn't ignore tables
|
|
if (field.reqd or field.fieldname in allow_fields) and field.fieldname not in exclude_fields:
|
|
if variant.get(field.fieldname) != item.get(field.fieldname):
|
|
if field.fieldtype == "Table":
|
|
variant.set(field.fieldname, [])
|
|
for d in item.get(field.fieldname):
|
|
row = copy.deepcopy(d)
|
|
if row.get("name"):
|
|
row.name = None
|
|
variant.append(field.fieldname, row)
|
|
else:
|
|
variant.set(field.fieldname, item.get(field.fieldname))
|
|
|
|
variant.variant_of = item.name
|
|
|
|
if "description" not in allow_fields:
|
|
if not variant.description:
|
|
variant.description = ""
|
|
else:
|
|
if item.variant_based_on == "Item Attribute":
|
|
if variant.attributes:
|
|
attributes_description = item.description or ""
|
|
for d in variant.attributes:
|
|
attributes_description += (
|
|
"<div>" + d.attribute + ": " + cstr(d.attribute_value) + "</div>"
|
|
)
|
|
|
|
if attributes_description not in (variant.description or ""):
|
|
variant.description = attributes_description
|
|
|
|
|
|
def make_variant_item_code(template_item_code, template_item_name, variant):
|
|
"""Uses template's item code and abbreviations to make variant's item code"""
|
|
if variant.item_code:
|
|
return
|
|
|
|
abbreviations = []
|
|
for attr in variant.attributes:
|
|
ia = frappe.qb.DocType("Item Attribute")
|
|
iav = frappe.qb.DocType("Item Attribute Value")
|
|
item_attribute = (
|
|
frappe.qb.from_(ia)
|
|
.left_join(iav)
|
|
.on(ia.name == iav.parent)
|
|
.select(ia.numeric_values, iav.abbr)
|
|
.where(
|
|
(ia.name == attr.attribute)
|
|
# attribute_value is a varchar column; cast the param to str so postgres doesn't choke on
|
|
# `varchar = numeric` for numeric attributes (where this side is irrelevant anyway, since
|
|
# numeric_values == 1 already satisfies the OR). Non-numeric values are already strings.
|
|
& ((iav.attribute_value == cstr(attr.attribute_value)) | (ia.numeric_values == 1))
|
|
)
|
|
.run(as_dict=True)
|
|
)
|
|
|
|
if not item_attribute:
|
|
continue
|
|
# frappe.throw(_('Invalid attribute {0} {1}').format(frappe.bold(attr.attribute),
|
|
# frappe.bold(attr.attribute_value)), title=_('Invalid Attribute'),
|
|
# exc=InvalidItemAttributeValueError)
|
|
|
|
abbr_or_value = (
|
|
cstr(attr.attribute_value) if item_attribute[0].numeric_values else item_attribute[0].abbr
|
|
)
|
|
abbreviations.append(abbr_or_value)
|
|
|
|
if abbreviations:
|
|
variant.item_code = "{}-{}".format(template_item_code, "-".join(abbreviations))
|
|
variant.item_name = "{}-{}".format(template_item_name, "-".join(abbreviations))
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_variant_doc_for_quick_entry(template: str, args: dict | str):
|
|
variant_based_on = frappe.db.get_value("Item", template, "variant_based_on")
|
|
args = frappe.parse_json(args)
|
|
if variant_based_on == "Manufacturer":
|
|
variant = get_variant(template, **args)
|
|
else:
|
|
existing_variant = get_variant(template, args)
|
|
if existing_variant:
|
|
return existing_variant
|
|
else:
|
|
variant = create_variant(template, args=args)
|
|
variant.name = variant.item_code
|
|
validate_item_variant_attributes(variant, args)
|
|
return variant.as_dict()
|