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 from erpnext.crm.doctype.lead.lead import make_customer from custom_ui.api.db.addresses import address_exists, create_address, create_address_links from custom_ui.api.db.contacts import check_and_get_contact, create_contact, create_contact_links # =============================================================================== # CLIENT MANAGEMENT API METHODS # =============================================================================== @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(categories) 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": [], "sales_orders": [], "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_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 = False tableRow = {} links = address.links customer_links = [link for link in links if link.link_doctype == "Customer"] if links else None customer_name = address.get("custom_customer_to_bill", None) 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: print("DEBUG: No customer links found and no customer to bill.") customer_name = "N/A" elif not customer_name and customer_links: print("DEBUG: No customer to bill. Customer links found:", customer_links) customer_name = frappe.get_value("Lead", customer_links[0].link_name, "custom_customer_name") if is_lead else customer_links[0].link_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']}" ) 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", []) # 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) if address_exists( data.get("address_line1"), data.get("address_line2"), data.get("city"), data.get("state"), data.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"), "company": data.get("company"), "custom_customer_name": customer_name, "customer_type": customer_type } 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 address creation address_doc = create_address({ "address_title": build_address_title(customer_name, data), "address_line1": data.get("address_line1"), "address_line2": data.get("address_line2"), "city": data.get("city"), "state": data.get("state"), "country": "United States", "pincode": data.get("pincode"), "customer_type": "Lead", "customer_name": client_doc.name }) #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 = create_contact({ "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, "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 }] }) contact_docs.append(contact_doc) # ##### Create links # # Customer -> Address # if client_doc.doctype == "Customer": # print("#####DEBUG: Linking address to customer.") # client_doc.append("custom_select_address", { # "address_name": address_doc.name, # }) # # Customer -> Contact # print("#####DEBUG: Linking contacts to customer.") # for contact_doc in contact_docs: # client_doc.append("custom_add_contacts", { # "contact": contact_doc.name, # "email": contact_doc.custom_email, # "phone": contact_doc.phone, # "role": contact_doc.role # }) # client_doc.save(ignore_permissions=True) # Address -> Customer/Lead create_address_links(address_doc, client_doc, contact_docs) # Contact -> Customer/Lead & Address create_contact_links(contact_docs, client_doc, address_doc) frappe.local.message_log = [] return build_success_response({ "customer": client_doc.as_dict(), "address": address_doc.as_dict(), "contacts": [contact_doc.as_dict() for contact_doc in contact_docs] }) 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.")