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 # =============================================================================== # 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.""" try: clientData = {"addresses": [], "contacts": [], "jobs": [], "sales_invoices": [], "payment_entries": [], "sales_orders": [], "tasks": []} customer = frappe.get_doc("Customer", client_name) clientData = {**clientData, **customer.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()) for address_link in customer.custom_select_address: address_doc = frappe.get_doc("Address", address_link.address_name) # # addressData = {"jobs": [], "contacts": []} # addressData = {**addressData, **address_doc.as_dict()} # addressData["estimates"] = frappe.db.get_all("Quotation", fields=["*"], filters={"custom_installation_address": address_doc.address_title}) # addressData["onsite_meetings"] = frappe.db.get_all("On-Site Meeting", fields=["*"], filters={"address": address_doc.address_title}) # jobs = frappe.db.get_all("Project", fields=["*"], or_filters=[ # ["custom_installation_address", "=", address.address_title], # ["custom_address", "=", address.address_title] # ]) # for job in jobs if jobs else []: # jobData = {} # jobData = {**jobData, **job} # jobData["sales_invoices"] = frappe.db.get_all("Sales Invoice", fields=["*"], filters={"project": job.name}) # jobData["payment_entries"] = frappe.db.get_all( # "Payment Entry", # fields=["*"], # filters={"party_type": "Customer"}, # or_filters=[ # ["party", "=", client_name], # ["party_name", "=", client_name] # ]) # jobData["sales_orders"] = frappe.db.get_all("Sales Order", fields=["*"], filters={"project": job.name}) # jobData["tasks"] = frappe.db.get_all("Task", fields=["*"], filters={"project": job.name}) # addressData["jobs"].append(jobData) clientData["addresses"].append(address_doc.as_dict()) 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: 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") 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 = 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["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 upsert_client(data): """Create or update a client (customer and address).""" try: data = json.loads(data) # Handle customer creation/update print("#####DEBUG: Upsert client data received:", data) print("#####DEBUG: Checking for existing customer with name:", data.get("customer_name")) customer = frappe.db.exists("Customer", {"customer_name": data.get("customer_name")}) if not customer: customer_doc = frappe.get_doc({ "doctype": "Customer", "customer_name": data.get("customer_name"), "customer_type": data.get("customer_type") }).insert(ignore_permissions=True) else: customer_doc = frappe.get_doc("Customer", data.get("customer_name")) print("Customer:", customer_doc.as_dict()) # Handle address creation print("#####DEBUG: Checking for existing address for customer:", data.get("customer_name")) existing_address = frappe.db.exists( "Address", { "address_line1": data.get("address_line1"), "city": data.get("city"), "state": data.get("state"), }) print("Existing address check:", existing_address) if existing_address: frappe.throw(f"Address already exists for customer {data.get('customer_name')}.", frappe.ValidationError) address_doc = frappe.get_doc({ "doctype": "Address", "address_title": data.get("address_title"), "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"), "custom_customer_to_bill": customer_doc.name }).insert(ignore_permissions=True) print("Address:", address_doc.as_dict()) #Handle contact creation contact_docs = [] for contact_data in data.get("contacts", []): if isinstance(contact_data, str): contact_data = json.loads(contact_data) print("#####DEBUG: Processing contact data:", contact_data) contact_exists = frappe.db.exists("Contact", {"email_id": contact_data.get("email"), "phone": contact_data.get("phone_number")}) if not contact_exists: is_primary_contact = 1 if contact_data.get("is_primary_contact") else 0 contact_doc = frappe.get_doc({ "doctype": "Contact", "first_name": contact_data.get("first_name"), "last_name": contact_data.get("last_name"), "email_id": contact_data.get("email"), "phone": contact_data.get("phone_number"), "custom_customer": customer_doc.name, "role": contact_data.get("contact_role", "Other"), "custom_email": contact_data.get("email"), "is_primary_contact": is_primary_contact }).insert(ignore_permissions=True) print("Created new contact:", contact_doc.as_dict()) else: contact_doc = frappe.get_doc("Contact", {"email_id": data.get("email")}) print("Contact already exists:", contact_doc.as_dict()) contact_docs.append(contact_doc) ##### Create links # Customer -> Address print("#####DEBUG: Creating links between customer, address, and contacts.") customer_doc.append("custom_select_address", { "address_name": address_doc.name, "address_line_1": address_doc.address_line1, "city": address_doc.city, "state": address_doc.state, "pincode": address_doc.pincode }) # Customer -> Contact print("#####DEBUG: Linking contacts to customer.") for contact_doc in contact_docs: print("Linking contact:", contact_doc.as_dict()) print("with role:", contact_doc.role) print("customer to append to:", customer_doc.as_dict()) customer_doc.append("custom_add_contacts", { "contact": contact_doc.name, "email": contact_doc.custom_email, "phone": contact_doc.phone, "role": contact_doc.role }) # Address -> Customer print("#####DEBUG: Linking address to customer.") address_doc.append("links", { "link_doctype": "Customer", "link_name": customer_doc.name }) # Address -> Contact print("#####DEBUG: Linking address to contacts.") address_doc.custom_contact = next((c.name for c in contact_docs if c.is_primary_contact), contact_docs[0].name) for contact_doc in contact_docs: address_doc.append("custom_linked_contacts", { "contact": contact_doc.name, "email": contact_doc.email_id, "phone": contact_doc.phone, "role": contact_doc.role }) # Contact -> Customer & Address print("#####DEBUG: Linking contacts to customer.") for contact_doc in contact_docs: contact_doc.append("links", { "link_doctype": "Customer", "link_name": customer_doc.name }) contact_doc.append("links", { "link_doctype": "Address", "link_name": address_doc.name }) contact_doc.custom_customer = customer_doc.name contact_doc.save(ignore_permissions=True) address_doc.save(ignore_permissions=True) customer_doc.save(ignore_permissions=True) return build_success_response({ "customer": customer_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", pluck="name") return build_success_response(client_names) except Exception as e: return build_error_response(str(e), 500)