diff --git a/erpnext/buying/doctype/supplier/supplier_import_provider.py b/erpnext/buying/doctype/supplier/supplier_import_provider.py new file mode 100644 index 00000000000..0a0f36c4ede --- /dev/null +++ b/erpnext/buying/doctype/supplier/supplier_import_provider.py @@ -0,0 +1,213 @@ +# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors +# License: GNU General Public License v3. See license.txt +"""Data Import provider for Supplier. + +Imports a Supplier together with Contact and Address records. Contact and Address are +separate doctypes linked through Dynamic Link rows; the provider exposes them as child +sections in import schema and creates linked records during row import. +""" + +import frappe +from frappe import _ + +from frappe.core.doctype.data_import.import_provider import ImportProvider +from frappe.core.doctype.data_import.importer import INSERT, UPDATE +from erpnext.selling.doctype.customer.mapper import parse_full_name + + +class SupplierImportProvider(ImportProvider): + def get_import_fields(self) -> dict: + return { + "fields": _doctype_docfields("Supplier"), + "child_tables": [ + *_doctype_child_tables("Supplier"), + { + "fieldname": "contacts", + "label": _("Contact"), + "fields": _contact_docfields(), + }, + { + "fieldname": "addresses", + "label": _("Address"), + "fields": _doctype_docfields("Address", prefer_plain_label=True), + }, + ], + } + + def validate(self, import_file) -> list[dict]: + warnings = [] + for payload in import_file.get_payloads_for_import(): + doc = payload.doc + row = payload.rows[0].row_number if payload.rows else None + + for contact in doc.get("contacts") or []: + # Keep parity with current Customer provider behavior: require at least one contact method. + if not (contact.get("email_id") or contact.get("mobile_no")): + missing = [ + label + for field, label in (("email_id", _("Email")), ("mobile_no", _("Mobile No"))) + if not contact.get(field) + ] + warnings.append( + { + "row": row, + "message": _("Contact in row {0} is missing {1}").format(row, ", ".join(missing)), + } + ) + + for address in doc.get("addresses") or []: + if not address.get("address_line1"): + continue + missing = [ + label + for field, label in (("city", _("City")), ("country", _("Country"))) + if not address.get(field) + ] + if missing: + warnings.append( + { + "row": row, + "message": _("Address in row {0} is missing {1}").format(row, ", ".join(missing)), + } + ) + + return warnings + + def import_row(self, importer, doc): + contact_rows = doc.pop("contacts", None) or [] + address_rows = doc.pop("addresses", None) or [] + has_child_rows = bool(contact_rows or address_rows) + supplier, import_action = self._persist_supplier(importer, doc, has_child_rows) + + self._create_contacts(supplier, contact_rows) + self._create_addresses(supplier, address_rows) + return supplier, import_action + + def _persist_supplier(self, importer, doc, has_child_rows): + if importer.import_type == INSERT: + return importer.insert_record(doc), None + + if importer.import_type == UPDATE: + return importer.update_record(doc, raise_if_no_changes=not has_child_rows), None + + return importer.upsert_record(doc) + + def _create_contacts(self, supplier, rows): + primary = None + for index, row in enumerate(rows): + row = dict(row) + email = row.pop("email_id", None) + mobile = row.pop("mobile_no", None) + flagged = frappe.utils.cint(row.pop("is_primary_contact", 0)) + + first_name, last_name, company_name = self._resolve_contact_names(supplier, row) + + contact_values = {k: v for k, v in row.items() if v not in (None, "")} + if first_name: + contact_values["first_name"] = first_name + if last_name: + contact_values["last_name"] = last_name + if company_name: + contact_values["company_name"] = company_name + + contact = frappe.get_doc( + { + "doctype": "Contact", + **contact_values, + "links": [{"link_doctype": "Supplier", "link_name": supplier.name}], + } + ) + if email: + contact.add_email(email, is_primary=True) + if mobile: + contact.add_phone(mobile, is_primary_mobile_no=True) + contact.insert() + if flagged or (primary is None and index == 0): + primary = contact + if primary: + frappe.db.set_value("Contact", primary.name, "is_primary_contact", 1) + supplier.db_set("supplier_primary_contact", primary.name) + supplier.db_set("mobile_no", primary.mobile_no) + supplier.db_set("email_id", primary.email_id) + + def _resolve_contact_names(self, supplier, row): + first_name = row.pop("first_name", None) + last_name = row.pop("last_name", None) + company_name = row.pop("company_name", None) + supplier_get = getattr(supplier, "get", None) + + def get_supplier_value(fieldname): + if callable(supplier_get): + return supplier_get(fieldname) + return getattr(supplier, fieldname, None) + + if supplier.supplier_type == "Individual": + first_name = first_name or get_supplier_value("first_name") + last_name = last_name or get_supplier_value("last_name") + if not first_name and supplier.supplier_name: + parsed_first, _, parsed_last = parse_full_name(supplier.supplier_name) + first_name = parsed_first + last_name = last_name or parsed_last + + return first_name, last_name, company_name + + def _create_addresses(self, supplier, rows): + from frappe.contacts.doctype.address.address import get_address_display + + primary = None + for index, row in enumerate(rows): + row = dict(row) + flagged = frappe.utils.cint(row.pop("is_primary_address", 0)) + if not row.get("address_line1"): + continue + row["address_type"] = row.get("address_type") or "Billing" + row["address_title"] = row.get("address_title") or supplier.supplier_name + address = frappe.get_doc( + { + "doctype": "Address", + **{k: v for k, v in row.items() if v not in (None, "")}, + "links": [{"link_doctype": "Supplier", "link_name": supplier.name}], + } + ) + address.insert() + if flagged or (primary is None and index == 0): + primary = address + if primary: + frappe.db.set_value("Address", primary.name, "is_primary_address", 1) + supplier.db_set("supplier_primary_address", primary.name) + supplier.db_set("primary_address", get_address_display(primary.name)) + + +def _doctype_docfields(doctype: str, prefer_plain_label: bool = False) -> list[dict]: + from frappe.model import display_fieldtypes, no_value_fields + + fields = [] + for df in frappe.get_meta(doctype).fields: + if df.fieldtype in no_value_fields or df.fieldtype in display_fieldtypes: + continue + if df.fieldname in ("lft", "rgt") or df.get("is_virtual"): + continue + field_dict = df.as_dict() + if prefer_plain_label: + field_dict["prefer_plain_label"] = 1 + fields.append(field_dict) + return fields + + +def _contact_docfields() -> list[dict]: + fields = _doctype_docfields("Contact", prefer_plain_label=True) + for field in fields: + if field.get("fieldname") == "email_id": + field["import_labels"] = ["Email ID"] + return fields + + +def _doctype_child_tables(doctype: str) -> list[dict]: + return [ + { + "fieldname": tf.fieldname, + "label": _(tf.label or tf.fieldname), + "fields": _doctype_docfields(tf.options), + } + for tf in frappe.get_meta(doctype).get_table_fields() + ]