823 lines
No EOL
38 KiB
Python
823 lines
No EOL
38 KiB
Python
import frappe, json
|
|
from custom_ui.db_utils import build_error_response, process_query_conditions, build_datatable_dict, get_count_or_filters, build_success_response, map_lead_client, build_address_title, normalize_name
|
|
from erpnext.crm.doctype.lead.lead import make_customer
|
|
from custom_ui.api.db.addresses import address_exists
|
|
from custom_ui.api.db.contacts import check_and_get_contact, create_contact, create_contact_links
|
|
from custom_ui.services import AddressService, ContactService, ClientService
|
|
|
|
# ===============================================================================
|
|
# CLIENT MANAGEMENT API METHODS
|
|
# ===============================================================================
|
|
|
|
@frappe.whitelist()
|
|
def add_addresses_contacts(client_name, company_name, addresses=[], contacts=[]):
|
|
if isinstance(addresses, str):
|
|
addresses = json.loads(addresses)
|
|
if isinstance(contacts, str):
|
|
contacts = json.loads(contacts)
|
|
print(f"DEBUG: add_addresses_contacts called with client_name: {client_name}, addresses: {addresses}, contacts: {contacts}")
|
|
try:
|
|
client_doc = ClientService.get_client_or_throw(client_name)
|
|
if contacts:
|
|
contact_docs = [frappe.get_doc("Contact", contact.contact) for contact in client_doc.contacts]
|
|
for contact in contacts:
|
|
contact_doc = None
|
|
if frappe.db.exists("Contact", {"email_id": contact.get("email")}):
|
|
contact_doc = frappe.get_doc("Contact", {"email_id": contact.get("email")})
|
|
else:
|
|
contact_doc = ContactService.create({
|
|
"first_name": contact.get("first_name"),
|
|
"last_name": contact.get("last_name"),
|
|
"email_id": contact.get("email"),
|
|
"role": contact.get("role"),
|
|
"phone": contact.get("phone"),
|
|
"custom_email": contact.get("email"),
|
|
"is_primary_contact": 0,
|
|
"customer_type": client_doc.doctype,
|
|
"customer_name": client_doc.name,
|
|
"email_ids": [{
|
|
"email": contact.get("email"),
|
|
"is_primary": 1
|
|
}],
|
|
"phone_nos": [{
|
|
"phone": contact.get("phone"),
|
|
"is_primary_phone": 1,
|
|
"is_primary_mobile_no": 1
|
|
}]
|
|
})
|
|
contact_doc.insert()
|
|
ClientService.append_link_v2(client_doc.name, "contacts", {"contact": contact_doc.name})
|
|
ContactService.link_contact_to_customer(contact_doc, client_doc.doctype, client_doc.name)
|
|
contact_docs.append(contact_doc)
|
|
address_docs = [frappe.get_doc("Address", link.address) for link in client_doc.properties]
|
|
for address in addresses:
|
|
address_doc = None
|
|
if frappe.db.exists("Address", {
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"city": address.get("city"),
|
|
# "state": address.get("state"),
|
|
"pincode": address.get("pincode")
|
|
}):
|
|
address_doc = frappe.get_doc("Address", {
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"city": address.get("city"),
|
|
# "state": address.get("state"),
|
|
"pincode": address.get("pincode")
|
|
})
|
|
else:
|
|
address_doc = AddressService.create({
|
|
"address_title": AddressService.build_address_title(customer_name=client_name, address_data=address),
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"city": address.get("city"),
|
|
"state": address.get("state"),
|
|
"pincode": address.get("pincode"),
|
|
"country": "United States",
|
|
"address_type": "Service",
|
|
"custom_billing_address": 0,
|
|
"is_primary_address": 0,
|
|
"is_service_address": 1,
|
|
"customer_type": client_doc.doctype,
|
|
"customer_name": client_doc.name
|
|
})
|
|
address_doc.insert()
|
|
if company_name not in [company.company for company in address_doc.companies]:
|
|
address_doc.append("companies", {"company": company_name})
|
|
address_doc.save(ignore_permissions=True)
|
|
AddressService.link_address_to_customer(address_doc, client_doc.doctype, client_doc.name)
|
|
for contact_to_link_idx in address.get("contacts", []):
|
|
contact_doc = contact_docs[contact_to_link_idx]
|
|
AddressService.link_address_to_contact(address_doc, contact_doc.name)
|
|
ContactService.link_contact_to_address(contact_doc, address_doc.name)
|
|
primary_contact = contact_docs[address.get("primary_contact", 0)]
|
|
AddressService.set_primary_contact(address_doc.name, primary_contact.name)
|
|
ClientService.append_link_v2(client_doc.name, "properties", {"address": address_doc.name})
|
|
address_docs.append(address_doc)
|
|
|
|
return build_success_response({
|
|
"contacts": [contact.as_dict() for contact in contact_docs],
|
|
"addresses": [address.as_dict() for address in address_docs],
|
|
"message": "Addresses and contacts added successfully."
|
|
})
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def check_client_exists(client_name):
|
|
"""Check if a client exists as either a Customer or a Lead.
|
|
Additionally, return a list of potential matches based on the client name."""
|
|
print("DEBUG: check_client_exists called with client_name:", client_name)
|
|
try:
|
|
exact_customer_match = frappe.db.exists("Customer", client_name)
|
|
exact_lead_match = frappe.db.exists("Lead", {"custom_customer_name": client_name})
|
|
customer_matches = frappe.get_all("Customer", pluck="name", filters={"name": ["like", f"%{client_name}%"]})
|
|
lead_matches = frappe.get_all("Lead", pluck="custom_customer_name", filters={"custom_customer_name": ["like", f"%{client_name}%"]})
|
|
# remove duplicates from potential matches between customers and leads
|
|
|
|
return build_success_response({
|
|
"exact_match": exact_customer_match or exact_lead_match,
|
|
"potential_matches": list(set(customer_matches + lead_matches))
|
|
})
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def get_client_status_counts(weekly=False, week_start_date=None, week_end_date=None):
|
|
"""Get counts of clients by status categories with optional weekly filtering."""
|
|
# Build base filters for date range if weekly filtering is enabled
|
|
# try:
|
|
# base_filters = {}
|
|
# if weekly and week_start_date and week_end_date:
|
|
# # Assuming you have a date field to filter by - adjust the field name as needed
|
|
# # Common options: creation, modified, custom_date_field, etc.
|
|
# base_filters["creation"] = ["between", [week_start_date, week_end_date]]
|
|
|
|
# # Helper function to merge base filters with status filters
|
|
# def get_filters(status_field, status_value):
|
|
# filters = {status_field: status_value}
|
|
# filters.update(base_filters)
|
|
# return filters
|
|
|
|
# onsite_meeting_scheduled_status_counts = {
|
|
# "label": "On-Site Meeting Scheduled",
|
|
# "not_started": frappe.db.count("Address", filters=get_filters("custom_onsite_meeting_scheduled", "Not Started")),
|
|
# "in_progress": frappe.db.count("Address", filters=get_filters("custom_onsite_meeting_scheduled", "In Progress")),
|
|
# "completed": frappe.db.count("Address", filters=get_filters("custom_onsite_meeting_scheduled", "Completed"))
|
|
# }
|
|
|
|
# estimate_sent_status_counts = {
|
|
# "label": "Estimate Sent",
|
|
# "not_started": frappe.db.count("Address", filters=get_filters("custom_estimate_sent_status", "Not Started")),
|
|
# "in_progress": frappe.db.count("Address", filters=get_filters("custom_estimate_sent_status", "In Progress")),
|
|
# "completed": frappe.db.count("Address", filters=get_filters("custom_estimate_sent_status", "Completed"))
|
|
# }
|
|
|
|
# job_status_counts = {
|
|
# "label": "Job Status",
|
|
# "not_started": frappe.db.count("Address", filters=get_filters("custom_job_status", "Not Started")),
|
|
# "in_progress": frappe.db.count("Address", filters=get_filters("custom_job_status", "In Progress")),
|
|
# "completed": frappe.db.count("Address", filters=get_filters("custom_job_status", "Completed"))
|
|
# }
|
|
|
|
# payment_received_status_counts = {
|
|
# "label": "Payment Received",
|
|
# "not_started": frappe.db.count("Address", filters=get_filters("custom_payment_received_status", "Not Started")),
|
|
# "in_progress": frappe.db.count("Address", filters=get_filters("custom_payment_received_status", "In Progress")),
|
|
# "completed": frappe.db.count("Address", filters=get_filters("custom_payment_received_status", "Completed"))
|
|
# }
|
|
|
|
# status_dicts = [
|
|
# onsite_meeting_scheduled_status_counts,
|
|
# estimate_sent_status_counts,
|
|
# job_status_counts,
|
|
# payment_received_status_counts
|
|
# ]
|
|
|
|
# categories = []
|
|
# for status_dict in status_dicts:
|
|
# category = {
|
|
# "label": status_dict["label"],
|
|
# "statuses": [
|
|
# {
|
|
# "color": "red",
|
|
# "label": "Not Started",
|
|
# "count": status_dict["not_started"]
|
|
# },
|
|
# {
|
|
# "color": "yellow",
|
|
# "label": "In Progress",
|
|
# "count": status_dict["in_progress"]
|
|
# },
|
|
# {
|
|
# "color": "green",
|
|
# "label": "Completed",
|
|
# "count": status_dict["completed"]
|
|
# }
|
|
# ]
|
|
# }
|
|
# categories.append(category)
|
|
|
|
return build_success_response("success")
|
|
# except frappe.ValidationError as ve:
|
|
# return build_error_response(str(ve), 400)
|
|
# except Exception as e:
|
|
# return build_error_response(str(e), 500)
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_client(client_name):
|
|
"""Get detailed information for a specific client including address, customer, and projects."""
|
|
print("DEBUG: get_client called with client_name:", client_name)
|
|
try:
|
|
clientData = {"addresses": [], "contacts": [], "jobs": [], "sales_invoices": [], "payment_entries": [], "tasks": []}
|
|
customer = check_and_get_client_doc(client_name)
|
|
if not customer:
|
|
return build_error_response(f"Client with name '{client_name}' does not exist.", 404)
|
|
print("DEBUG: Retrieved customer/lead document:", customer.as_dict())
|
|
clientData = {**clientData, **customer.as_dict()}
|
|
if customer.doctype == "Lead":
|
|
clientData.update(map_lead_client(clientData))
|
|
links = []
|
|
if customer.doctype == "Customer":
|
|
for address_link in customer.custom_select_address:
|
|
address_doc = frappe.get_doc("Address", address_link.address_name)
|
|
clientData["addresses"].append(address_doc.as_dict())
|
|
for contact_link in customer.custom_add_contacts:
|
|
contact_doc = frappe.get_doc("Contact", contact_link.contact)
|
|
clientData["contacts"].append(contact_doc.as_dict())
|
|
|
|
else:
|
|
links = frappe.get_all(
|
|
"Dynamic Link",
|
|
filters={
|
|
"link_doctype": "Lead",
|
|
"link_name": customer.name,
|
|
"parenttype": ["in", ["Address", "Contact"]],
|
|
},
|
|
fields=[
|
|
"parenttype as link_doctype",
|
|
"parent as link_name",
|
|
]
|
|
)
|
|
print("DEBUG: Retrieved links from lead:", links)
|
|
for link in links:
|
|
print("DEBUG: Processing link:", link)
|
|
linked_doc = frappe.get_doc(link["link_doctype"], link["link_name"])
|
|
if link["link_doctype"] == "Contact":
|
|
clientData["contacts"].append(linked_doc.as_dict())
|
|
elif link["link_doctype"] == "Address":
|
|
clientData["addresses"].append(linked_doc.as_dict())
|
|
|
|
# TODO: Continue getting other linked docs like jobs, invoices, etc.
|
|
print("DEBUG: Final client data prepared:", clientData)
|
|
return build_success_response(clientData)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def get_client_v2(client_name):
|
|
"""Get detailed information for a specific client including address, customer, and projects."""
|
|
print("DEBUG: get_client_v2 called with client_name:", client_name)
|
|
try:
|
|
clientData = {"addresses": [], "jobs": [], "payment_entries": [], "tasks": []}
|
|
customer = check_and_get_client_doc(client_name)
|
|
if not customer:
|
|
return build_error_response(f"Client with name '{client_name}' does not exist.", 404)
|
|
print("DEBUG: Retrieved customer/lead document:", customer.as_dict())
|
|
clientData = {**clientData, **customer.as_dict()}
|
|
clientData["contacts"] = [ContactService.get_or_throw(link.contact) for link in clientData["contacts"]]
|
|
clientData["addresses"] = [AddressService.get_or_throw(link.address) for link in clientData["properties"]]
|
|
if clientData["doctype"] == "Lead":
|
|
clientData["customer_name"] = customer.custom_customer_name
|
|
|
|
# TODO: Continue getting other linked docs like jobs, invoices, etc.
|
|
print("DEBUG: Final client data prepared:", clientData)
|
|
return build_success_response(clientData)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_clients_table_data_v2(filters={}, sortings=[], page=1, page_size=10):
|
|
"""Get paginated client table data with filtering and sorting support."""
|
|
try:
|
|
filters = json.loads(filters) if isinstance(filters, str) else filters
|
|
sortings = json.loads(sortings) if isinstance(sortings, str) else sortings
|
|
page = int(page)
|
|
page_size = int(page_size)
|
|
print("DEBUG: Raw client table query received:", {
|
|
"filters": filters,
|
|
"sortings": sortings,
|
|
"page": page,
|
|
"page_size": page_size
|
|
})
|
|
where_clauses = []
|
|
values = []
|
|
if filters.get("company"):
|
|
where_clauses.append("c.company = %s")
|
|
values.append(filters["company"]["value"])
|
|
|
|
if filters.get("address"):
|
|
where_clauses.append("a.full_address LIKE %s")
|
|
values.append(f"%{filters['address']['value']}%")
|
|
|
|
if filters.get("customer_name"):
|
|
where_clauses.append("a.customer_name LIKE %s")
|
|
values.append(f"%{filters['customer_name']['value']}%")
|
|
|
|
where_sql = ""
|
|
if where_clauses:
|
|
where_sql = "WHERE " + " AND ".join(where_clauses)
|
|
|
|
offset = (page - 1) * page_size
|
|
print("DEBUG: Constructed SQL where clause:", where_sql)
|
|
address_names = frappe.db.sql(f"""
|
|
SELECT DISTINCT a.name
|
|
FROM `tabAddress` a
|
|
LEFT JOIN `tabAddress Company Link` c ON c.parent = a.name
|
|
{where_sql}
|
|
ORDER BY a.modified DESC
|
|
LIMIT %s OFFSET %s
|
|
""", values + [page_size, offset], as_dict=True)
|
|
print("DEBUG: Address names retrieved:", address_names)
|
|
|
|
count = frappe.db.sql(f"""
|
|
SELECT COUNT(DISTINCT a.name) as count
|
|
FROM `tabAddress` a
|
|
LEFT JOIN `tabAddress Company Link` c ON c.parent = a.name
|
|
{where_sql}
|
|
""", values, as_dict=True)[0]["count"]
|
|
tableRows = []
|
|
for address_name in address_names:
|
|
address = AddressService.get_or_throw(address_name["name"])
|
|
tableRow = {}
|
|
tableRow["id"] = address.name
|
|
tableRow["address"] = address.full_address
|
|
tableRow["client_type"] = address.customer_type
|
|
tableRow["customer_name"] = normalize_name(address.customer_name, "-#-")
|
|
tableRow["companies"] = ", ".join([link.company for link in address.get("companies", [])])
|
|
tableRows.append(tableRow)
|
|
|
|
table_data = build_datatable_dict(data=tableRows, count=count, page=page, page_size=page_size)
|
|
|
|
return build_success_response(table_data)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
print("ERROR in get_clients_table_data_v2:", str(e))
|
|
return build_error_response(str(e), 500)
|
|
|
|
|
|
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_clients_table_data(filters={}, sortings=[], page=1, page_size=10):
|
|
"""Get paginated client table data with filtering and sorting support."""
|
|
try:
|
|
|
|
print("DEBUG: Raw client table query received:", {
|
|
"filters": filters,
|
|
"sortings": sortings,
|
|
"page": page,
|
|
"page_size": page_size
|
|
})
|
|
|
|
processed_filters, processed_sortings, is_or, page, page_size = process_query_conditions(filters, sortings, page, page_size)
|
|
print("DEBUG: Processed filters:", processed_filters)
|
|
print("DEBUG: Processed sortings:", processed_sortings)
|
|
# Handle count with proper OR filter support
|
|
if is_or:
|
|
count = frappe.db.sql(*get_count_or_filters("Address", processed_filters))[0][0]
|
|
else:
|
|
count = frappe.db.count("Address", filters=processed_filters)
|
|
|
|
print("DEBUG: Count of addresses matching filters:", count)
|
|
|
|
address_names = frappe.db.get_all(
|
|
"Address",
|
|
fields=["name"],
|
|
filters=processed_filters if not is_or else None,
|
|
or_filters=processed_filters if is_or else None,
|
|
limit=page_size,
|
|
start=(page - 1) * page_size,
|
|
order_by=processed_sortings
|
|
)
|
|
|
|
addresses = [frappe.get_doc("Address", addr["name"]).as_dict() for addr in address_names]
|
|
tableRows = []
|
|
for address in addresses:
|
|
is_lead = address.customer_type == "Lead"
|
|
print("##########IS LEAD:", is_lead)
|
|
tableRow = {}
|
|
links = address.links
|
|
# customer_links = [link for link in links if link.link_doctype == "Customer"] if links else None
|
|
customer_name = address.get("customer_name")
|
|
# if not customer_links:
|
|
# customer_links = [link for link in links if link.link_doctype == "Lead"] if links else None
|
|
# is_lead = True if customer_links else False
|
|
# if not customer_name and not customer_links:
|
|
# customer_name = frappe.get_value("Lead", address.get("customer_name"), "custom_customer_name")
|
|
if is_lead:
|
|
# print("DEBUG: No customer to bill. Customer links found:", customer_links)
|
|
customer_name = frappe.get_value("Lead", address.get("customer_name"), "custom_customer_name")
|
|
tableRow["id"] = address["name"]
|
|
tableRow["customer_name"] = customer_name
|
|
tableRow["address"] = (
|
|
f"{address['address_line1']}"
|
|
f"{' ' + address['address_line2'] if address['address_line2'] else ''} "
|
|
f"{address['city']}, {address['state']} {address['pincode']}"
|
|
)
|
|
print("########IS LEAD @TABLE ROW:", is_lead)
|
|
tableRow["client_type"] = "Lead" if is_lead else "Customer"
|
|
# tableRow["appointment_scheduled_status"] = address.custom_onsite_meeting_scheduled
|
|
# tableRow["estimate_sent_status"] = address.custom_estimate_sent_status
|
|
# tableRow["job_status"] = address.custom_job_status
|
|
tableRow["payment_received_status"] = address.custom_payment_received_status
|
|
tableRows.append(tableRow)
|
|
tableDataDict = build_datatable_dict(data=tableRows, count=count, page=page, page_size=page_size)
|
|
return build_success_response(tableDataDict)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def update_client_info(client_name, data):
|
|
"""Update client information for a given client."""
|
|
try:
|
|
data = json.loads(data)
|
|
print("DEBUG: update_client_info called with client_name:", client_name, "and data:", data)
|
|
client_doc = check_and_get_client_doc(client_name)
|
|
if not client_doc:
|
|
return build_error_response(f"Client with name '{client_name}' does not exist.", 404)
|
|
address_updates = data.get("addresses", [])
|
|
contact_updates = data.get("contacts", [])
|
|
customer_updates = data.get("customer", {})
|
|
# Update addresses
|
|
if address_updates:
|
|
for addr_data in address_updates:
|
|
update_address(addr_data)
|
|
# Update contacts
|
|
if contact_updates:
|
|
for contact_data in contact_updates:
|
|
update_contact(contact_data)
|
|
# Update customer/lead
|
|
if customer_updates:
|
|
for field, value in customer_updates.items():
|
|
if hasattr(client_doc, field):
|
|
setattr(client_doc, field, value)
|
|
client_doc.save(ignore_permissions=True)
|
|
frappe.local.message_log = []
|
|
return get_client(client_name)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def upsert_client(data):
|
|
"""Create a client (customer and address)."""
|
|
try:
|
|
data = json.loads(data)
|
|
print("#####DEBUG: Create client data received:", data)
|
|
|
|
customer_name = data.get("customer_name")
|
|
contacts = data.get("contacts", [])
|
|
addresses = data.get("addresses", [])
|
|
|
|
# Check for existing address
|
|
client_doc = check_and_get_client_doc(customer_name)
|
|
if client_doc:
|
|
return build_error_response(f"Client with name '{customer_name}' already exists.", 400)
|
|
# for address in addresses:
|
|
# if address_exists(
|
|
# address.get("address_line1"),
|
|
# address.get("address_line2"),
|
|
# address.get("city"),
|
|
# address.get("state"),
|
|
# address.get("pincode")
|
|
# ):
|
|
# return build_error_response("This address already exists. Please use a different address or search for the address to find the associated client.", 400)
|
|
|
|
# Handle customer creation/update
|
|
|
|
print("#####DEBUG: Creating new lead.")
|
|
customer_type = data.get("customer_type", "Individual")
|
|
primary_contact = find_primary_contact_or_throw(contacts)
|
|
lead_data = {
|
|
"first_name": primary_contact.get("first_name"),
|
|
"last_name": primary_contact.get("last_name"),
|
|
"email_id": primary_contact.get("email"),
|
|
"phone": primary_contact.get("phone_number"),
|
|
"custom_customer_name": customer_name,
|
|
"customer_type": customer_type,
|
|
"companies": [{ "company": data.get("company_name")
|
|
}]
|
|
}
|
|
if customer_type == "Company":
|
|
lead_data["company_name"] = data.get("customer_name")
|
|
client_doc = create_lead(lead_data)
|
|
print(f"#####DEBUG: {client_doc.doctype}:", client_doc.as_dict())
|
|
|
|
#Handle contact creation
|
|
contact_docs = []
|
|
for contact_data in contacts:
|
|
if isinstance(contact_data, str):
|
|
contact_data = json.loads(contact_data)
|
|
print("#####DEBUG: Processing contact data:", contact_data)
|
|
contact_doc = check_and_get_contact(
|
|
contact_data.get("first_name"),
|
|
contact_data.get("last_name"),
|
|
contact_data.get("email"),
|
|
contact_data.get("phone_number")
|
|
)
|
|
if not contact_doc:
|
|
print("#####DEBUG: No existing contact found. Creating new contact.")
|
|
contact_doc = ContactService.create({
|
|
"first_name": contact_data.get("first_name"),
|
|
"last_name": contact_data.get("last_name"),
|
|
"role": contact_data.get("contact_role", "Other"),
|
|
"custom_email": contact_data.get("email"),
|
|
"is_primary_contact": 1 if contact_data.get("is_primary", False) else 0,
|
|
"customer_type": "Lead",
|
|
"customer_name": client_doc.name,
|
|
"email_ids": [{
|
|
"email_id": contact_data.get("email"),
|
|
"is_primary": 1
|
|
}],
|
|
"phone_nos": [{
|
|
"phone": contact_data.get("phone_number"),
|
|
"is_primary_mobile_no": 1,
|
|
"is_primary_phone": 1
|
|
}]
|
|
})
|
|
ContactService.link_contact_to_customer(contact_doc, "Lead", client_doc.name)
|
|
contact_docs.append(contact_doc)
|
|
|
|
# Link all contacts to client after creating them
|
|
client_doc.reload()
|
|
for idx, contact_data in enumerate(contacts):
|
|
if isinstance(contact_data, str):
|
|
contact_data = json.loads(contact_data)
|
|
contact_doc = contact_docs[idx]
|
|
client_doc.append("contacts", {
|
|
"contact": contact_doc.name
|
|
})
|
|
if contact_data.get("is_primary", False):
|
|
client_doc.primary_contact = contact_doc.name
|
|
client_doc.save(ignore_permissions=True)
|
|
|
|
# Handle address creation
|
|
address_docs = []
|
|
for address in addresses:
|
|
|
|
is_billing = True if address.get("is_billing_address") else False
|
|
is_service = True if address.get("is_service_address") else False
|
|
address_exists = frappe.db.exists("Address", {
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"city": address.get("city"),
|
|
"pincode": address.get("pincode")
|
|
})
|
|
address_doc = None
|
|
if address_exists:
|
|
address_doc = frappe.get_doc("Address", address_exists)
|
|
else:
|
|
print("#####DEBUG: Creating address with data:", address)
|
|
address_doc = AddressService.create_address({
|
|
"address_title": AddressService.build_address_title(customer_name, address),
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"address_type": "Billing" if is_billing else "Service",
|
|
"custom_billing_address": is_billing,
|
|
"is_service_address": is_service,
|
|
"is_primary_address": is_billing,
|
|
"city": address.get("city"),
|
|
"state": address.get("state"),
|
|
"country": "United States",
|
|
"pincode": address.get("pincode"),
|
|
"customer_type": "Lead",
|
|
"customer_name": client_doc.name,
|
|
"companies": [{ "company": data.get("company_name") }]
|
|
})
|
|
AddressService.link_address_to_customer(address_doc, "Lead", client_doc.name)
|
|
address_doc.reload()
|
|
if is_billing:
|
|
client_doc.custom_billing_address = address_doc.name
|
|
client_doc.save(ignore_permissions=True)
|
|
for contact_to_link_idx in address.get("contacts", []):
|
|
contact_doc = contact_docs[contact_to_link_idx]
|
|
AddressService.link_address_to_contact(address_doc, contact_doc.name)
|
|
address_doc.reload()
|
|
ContactService.link_contact_to_address(contact_doc, address_doc.name)
|
|
primary_contact = contact_docs[address.get("primary_contact)", 0)]
|
|
AddressService.set_primary_contact(address_doc.name, primary_contact.name)
|
|
address_docs.append(address_doc)
|
|
|
|
# Link all addresses to client after creating them
|
|
client_doc.reload()
|
|
for address_doc in address_docs:
|
|
client_doc.append("properties", {
|
|
"address": address_doc.name
|
|
})
|
|
client_doc.save(ignore_permissions=True)
|
|
client_dict = client_doc.as_dict()
|
|
client_dict["contacts"] = [contact.as_dict() for contact in contact_docs]
|
|
client_dict["addresses"] = [address.as_dict() for address in address_docs]
|
|
|
|
frappe.local.message_log = []
|
|
return build_success_response(client_dict)
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def get_client_names(search_term):
|
|
"""Search for client names matching the search term."""
|
|
try:
|
|
search_pattern = f"%{search_term}%"
|
|
client_names = frappe.db.get_all(
|
|
"Customer",
|
|
filters={"customer_name": ["like", search_pattern]},
|
|
pluck="name")
|
|
return build_success_response(client_names)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
def check_if_customer(client_name):
|
|
"""Check if the given client name corresponds to a Customer."""
|
|
return frappe.db.exists("Customer", client_name) is not None
|
|
|
|
def check_and_get_client_doc(client_name):
|
|
"""Check if a client exists as Customer or Lead and return the document."""
|
|
print("DEBUG: Checking for existing client with name:", client_name)
|
|
customer = None
|
|
if check_if_customer(client_name):
|
|
print("DEBUG: Client found as Customer.")
|
|
customer = frappe.get_doc("Customer", client_name)
|
|
else:
|
|
print("DEBUG: Client not found as Customer. Checking Lead.")
|
|
lead_name = frappe.db.get_all("Lead", pluck="name", filters={"company_name": client_name})
|
|
if not lead_name:
|
|
lead_name = frappe.db.get_all("Lead", pluck="name", filters={"lead_name": client_name})
|
|
if not lead_name:
|
|
lead_name = frappe.db.get_all("Lead", pluck="name", filters={"custom_customer_name": client_name})
|
|
if lead_name:
|
|
print("DEBUG: Client found as Lead.")
|
|
customer = frappe.get_doc("Lead", lead_name[0])
|
|
return customer
|
|
|
|
def convert_lead_to_customer(lead_name):
|
|
lead = frappe.get_doc("Lead", lead_name)
|
|
customer = make_customer(lead)
|
|
customer.insert(ignore_permissions=True)
|
|
|
|
|
|
def create_lead(lead_data):
|
|
lead = frappe.get_doc({
|
|
"doctype": "Lead",
|
|
**lead_data
|
|
})
|
|
lead.insert(ignore_permissions=True)
|
|
return lead
|
|
|
|
def get_customer_or_lead(client_name):
|
|
if check_if_customer(client_name):
|
|
return frappe.get_doc("Customer", client_name)
|
|
else:
|
|
lead_name = frappe.db.get_all("Lead", pluck="name", filters={"lead_name": client_name})[0]
|
|
return frappe.get_doc("Lead", lead_name)
|
|
|
|
def find_primary_contact_or_throw(contacts):
|
|
for contact in contacts:
|
|
if contact.get("is_primary"):
|
|
print("#####DEBUG: Primary contact found:", contact)
|
|
return contact
|
|
raise ValueError("No primary contact found in contacts list.")
|
|
|
|
|
|
def find_contact_in_list(contact_docs, contact_ref):
|
|
"""Find a contact document in a list by matching first_name, last_name, and email."""
|
|
if not isinstance(contact_ref, dict):
|
|
return None
|
|
ref_first = contact_ref.get("first_name", "")
|
|
ref_last = contact_ref.get("last_name", "")
|
|
ref_email = contact_ref.get("email", "")
|
|
for doc in contact_docs:
|
|
if (doc.first_name == ref_first and
|
|
doc.last_name == ref_last and
|
|
(doc.email_id == ref_email or doc.custom_email == ref_email)):
|
|
return doc
|
|
return None
|
|
|
|
|
|
@frappe.whitelist()
|
|
def create_client_contacts_addresses(client_name, company, contacts=[], addresses=[]):
|
|
"""Create or link contacts and addresses for an existing client.
|
|
|
|
If a contact or address already exists, it will be linked to the client
|
|
instead of creating a duplicate.
|
|
"""
|
|
if isinstance(contacts, str):
|
|
contacts = json.loads(contacts)
|
|
if isinstance(addresses, str):
|
|
addresses = json.loads(addresses)
|
|
print(f"DEBUG: create_client_contacts_addresses called with client_name: {client_name}, company: {company}")
|
|
try:
|
|
client_doc = ClientService.get_client_or_throw(client_name)
|
|
|
|
# Build list of existing client contacts (preserves frontend index order)
|
|
existing_contact_docs = [frappe.get_doc("Contact", link.contact) for link in client_doc.contacts]
|
|
|
|
# Process new contacts
|
|
new_contact_docs = []
|
|
for contact in contacts:
|
|
contact_doc = check_and_get_contact(
|
|
contact.get("first_name"),
|
|
contact.get("last_name"),
|
|
contact.get("email"),
|
|
contact.get("phone_number")
|
|
)
|
|
if not contact_doc:
|
|
contact_doc = ContactService.create({
|
|
"first_name": contact.get("first_name"),
|
|
"last_name": contact.get("last_name"),
|
|
"role": contact.get("contact_role", "Other"),
|
|
"custom_email": contact.get("email"),
|
|
"is_primary_contact": 1 if contact.get("is_primary") else 0,
|
|
"customer_type": client_doc.doctype,
|
|
"customer_name": client_doc.name,
|
|
"email_ids": [{
|
|
"email_id": contact.get("email"),
|
|
"is_primary": 1
|
|
}],
|
|
"phone_nos": [{
|
|
"phone": contact.get("phone_number"),
|
|
"is_primary_phone": 1,
|
|
"is_primary_mobile_no": 1
|
|
}]
|
|
})
|
|
ContactService.link_contact_to_customer(contact_doc, client_doc.doctype, client_doc.name)
|
|
ClientService.append_link_v2(client_doc.name, "contacts", {"contact": contact_doc.name})
|
|
new_contact_docs.append(contact_doc)
|
|
|
|
# Combined contact list: existing client contacts + newly created/linked contacts
|
|
# Address contact indices reference this combined list
|
|
all_contact_docs = existing_contact_docs + new_contact_docs
|
|
|
|
# Process addresses
|
|
address_docs = []
|
|
for address in addresses:
|
|
filters = {
|
|
"address_line1": address.get("address_line1"),
|
|
"city": address.get("city"),
|
|
"pincode": address.get("pincode")
|
|
}
|
|
if address.get("address_line2"):
|
|
filters["address_line2"] = address.get("address_line2")
|
|
|
|
existing_address = frappe.db.exists("Address", filters)
|
|
if existing_address:
|
|
address_doc = frappe.get_doc("Address", existing_address)
|
|
else:
|
|
address_doc = AddressService.create({
|
|
"address_title": AddressService.build_address_title(client_name, address),
|
|
"address_line1": address.get("address_line1"),
|
|
"address_line2": address.get("address_line2"),
|
|
"city": address.get("city"),
|
|
"state": address.get("state"),
|
|
"pincode": address.get("pincode"),
|
|
"country": "United States",
|
|
"address_type": "Service",
|
|
"custom_billing_address": 0,
|
|
"is_primary_address": 0,
|
|
"is_service_address": 1,
|
|
"customer_type": client_doc.doctype,
|
|
"customer_name": client_doc.name
|
|
})
|
|
|
|
# Add company if not already present
|
|
if company not in [c.company for c in address_doc.companies]:
|
|
address_doc.append("companies", {"company": company})
|
|
address_doc.save(ignore_permissions=True)
|
|
|
|
# Link address to customer
|
|
AddressService.link_address_to_customer(address_doc, client_doc.doctype, client_doc.name)
|
|
|
|
# Link selected contacts to address
|
|
for contact_ref in address.get("contacts", []):
|
|
if not contact_ref:
|
|
continue
|
|
# Contact references are dicts with first_name, last_name, email
|
|
contact_doc = find_contact_in_list(all_contact_docs, contact_ref)
|
|
if contact_doc:
|
|
AddressService.link_address_to_contact(address_doc, contact_doc.name)
|
|
ContactService.link_contact_to_address(contact_doc, address_doc.name)
|
|
|
|
# Set primary contact for address
|
|
primary_ref = address.get("primary_contact")
|
|
if primary_ref:
|
|
primary_contact = find_contact_in_list(all_contact_docs, primary_ref)
|
|
if primary_contact:
|
|
AddressService.set_primary_contact(address_doc.name, primary_contact.name)
|
|
|
|
# Link address to client
|
|
ClientService.append_link_v2(client_doc.name, "properties", {"address": address_doc.name})
|
|
address_docs.append(address_doc)
|
|
|
|
return build_success_response({
|
|
"contacts": [c.as_dict() for c in new_contact_docs],
|
|
"addresses": [a.as_dict() for a in address_docs],
|
|
"message": "Contacts and addresses created/linked successfully."
|
|
})
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500) |