feat: add Customer import providers for data import functionality

erpnext/hooks.py

@@ -752,3 +752,13 @@ repost_allowed_doctypes = [
 	"Payment Entry",
 	"Purchase Receipt",
 ]
+
+
+# Data Import
+# -----------
+# Custom Import Providers plug provider-owned field schema, validation and import logic into
+# the standard Data Import. See frappe.core.doctype.data_import.import_provider.
+data_import_providers = {
+	"Customer": "erpnext.selling.doctype.customer.customer_import_provider.CustomerImportProvider",
+	"Supplier": "erpnext.buying.doctype.supplier.supplier_import_provider.SupplierImportProvider",
+}

erpnext/selling/doctype/customer/customer_import_provider.py

@@ -0,0 +1,240 @@
+# Copyright (c) 2024, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+"""Data Import provider for Customer.
+
+Imports a Customer together with its Contact and Address records. In the schema Contact and
+Address are separate doctypes linked to the Customer via a Dynamic Link; the provider exposes
+them as child tables in the import and wires up the links + primary flags in ``import_row``.
+"""
+
+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 .mapper import parse_full_name
+
+
+class CustomerImportProvider(ImportProvider):
+	def get_import_fields(self) -> dict:
+		"""Field schema for the picker: a Customer with its own child tables, plus Contact and
+		Address (linked doctypes exposed as extra child tables).
+
+		For Customer the columns ARE those doctypes' fields, so we read complete docfields from
+		meta. (This is the provider's choice — Data Import itself never reads meta here.)
+		"""
+		return {
+			"fields": _doctype_docfields("Customer"),
+			"child_tables": [
+				*_doctype_child_tables("Customer"),
+				{
+					"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]:
+		"""Business checks on top of the framework's basic Select/Link/Date validation."""
+		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 []:
+				# TODO: Need to Confirm if need to throw Error is Contact First Name not present. For now not throwing Warning.
+				# first_name, _last_name, company_name = self._resolve_contact_names(doc, dict(contact))
+				# if (contact.get("email_id") or contact.get("mobile_no")) and not (first_name or company_name):
+				if not (contact.get("email_id") or contact.get("mobile_no")):
+					# warnings.append(
+					# 	{
+					# 		"row": row,
+					# 		"message": _("Contact in row {0} is missing a First Name").format(
+					# 			row
+					# 		),
+					# 	}
+					# )
+					# Report exactly which fields are missing
+					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):
+		"""Persist Customer per Import Type, then create linked Contacts/Addresses."""
+		contact_rows = doc.pop("contacts", None) or []
+		address_rows = doc.pop("addresses", None) or []
+		has_child_rows = bool(contact_rows or address_rows)
+		customer, import_action = self._persist_customer(importer, doc, has_child_rows)
+
+		self._create_contacts(customer, contact_rows)
+		self._create_addresses(customer, address_rows)
+		return customer, import_action
+
+	def _persist_customer(self, importer, doc, has_child_rows):
+		"""Use the core Importer paths so provider imports honor Insert/Update/Upsert semantics."""
+		if importer.import_type == INSERT:
+			return importer.insert_record(doc), None
+
+		if importer.import_type == UPDATE:
+			# Allow child-only updates (e.g. add Contact/Address) without failing on unchanged Customer fields.
+			return importer.update_record(doc, raise_if_no_changes=not has_child_rows), None
+
+		return importer.upsert_record(doc)
+
+	def _create_contacts(self, customer, 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(customer, row)
+
+			# TODO: Check if first_name or company_name is required for Contact creation. For now lets create Contact even if both are missing, as long as email or mobile is present.
+			# if not first_name and not company_name:
+			# 	continue
+
+			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": "Customer", "link_name": customer.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)
+			customer.db_set("customer_primary_contact", primary.name)
+			customer.db_set("mobile_no", primary.mobile_no)
+			customer.db_set("email_id", primary.email_id)
+
+	def _resolve_contact_names(self, customer, row):
+		"""Resolve Contact names from row data and fall back to Customer data when needed."""
+		first_name = row.pop("first_name", None)
+		last_name = row.pop("last_name", None)
+		company_name = row.pop("company_name", None)
+		customer_get = getattr(customer, "get", None)
+
+		def get_customer_value(fieldname):
+			if callable(customer_get):
+				return customer_get(fieldname)
+			return getattr(customer, fieldname, None)
+
+		if customer.customer_type == "Individual":
+			first_name = first_name or get_customer_value("first_name")
+			last_name = last_name or get_customer_value("last_name")
+			if not first_name and customer.customer_name:
+				parsed_first, _, parsed_last = parse_full_name(customer.customer_name)
+				first_name = parsed_first
+				last_name = last_name or parsed_last
+
+		return first_name, last_name, company_name
+
+	def _create_addresses(self, customer, 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 customer.customer_name
+			address = frappe.get_doc(
+				{
+					"doctype": "Address",
+					**{k: v for k, v in row.items() if v not in (None, "")},
+					"links": [{"link_doctype": "Customer", "link_name": customer.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)
+			customer.db_set("customer_primary_address", primary.name)
+			customer.db_set("primary_address", get_address_display(primary.name))
+
+
+def _doctype_docfields(doctype: str, prefer_plain_label: bool = False) -> list[dict]:
+	"""Non-table importable fields of ``doctype`` as complete docfield dicts."""
+	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]:
+	"""Contact fields with plain-label and import-header aliases for common CSV headers."""
+	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]:
+	"""``doctype``'s own child tables as schema groups (fieldname, label, fields)."""
+	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()
+	]