410 lines
20 KiB
Python
410 lines
20 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
|
|
|
|
# ===============================================================================
|
|
# 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": []}
|
|
client_exists = frappe.db.exists("Customer", client_name)
|
|
if client_exists:
|
|
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={"lead_name": client_name})[0]
|
|
customer = frappe.get_doc("Lead", lead_name)
|
|
clientData = {**clientData, **customer.as_dict()}
|
|
if customer.doctype == "Customer":
|
|
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:
|
|
contact_names = frappe.db.get_all("Contact", pluck="name", filters={"links": ["like", f'%{{"link_doctype": "Lead", "link_name": client_name}}%']})
|
|
for contact_name in contact_names:
|
|
contact_doc = frappe.get_doc("Contact", contact_name)
|
|
clientData["contacts"].append(contact_doc.as_dict())
|
|
|
|
if customer.doctype == "Customer":
|
|
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())
|
|
else:
|
|
address_names = frappe.db.get_all("Address", pluck="name", filters={"links": ["like", f'%{{"link_doctype": "Lead", "link_name": client_name}}%']})
|
|
for address_name in address_names:
|
|
address_doc = frappe.get_doc("Address", address_name)
|
|
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:
|
|
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, "lead_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["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:
|
|
print("#####DEBUG: No existing customer found. Checking for existing lead")
|
|
customer = frappe.db.exists("Lead", {"lead_name": data.get("customer_name")})
|
|
else:
|
|
print("#####DEBUG: Existing customer found:", customer)
|
|
|
|
if not customer:
|
|
print("#####DEBUG: No existing lead found. Creating new lead.")
|
|
is_individual = data.get("customer_type") == "Individual"
|
|
|
|
primary_contact = next((c for c in data.get("contacts", []) if c.get("is_primary")), None)
|
|
if not primary_contact:
|
|
return build_error_response("Primary contact information is required to create a new customer.", 400)
|
|
print("#####DEBUG: Primary contact found:", primary_contact)
|
|
|
|
new_lead_data = {
|
|
"doctype": "Lead",
|
|
"lead_name": data.get("customer_name"),
|
|
"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"),
|
|
"customer_type": data.get("customer_type"),
|
|
"company": data.get("company")
|
|
}
|
|
print("#####DEBUG: New lead data prepared:", new_lead_data)
|
|
new_client_doc = frappe.get_doc(new_lead_data).insert(ignore_permissions=True)
|
|
else:
|
|
new_client_doc = frappe.get_doc("Customer", data.get("customer_name"))
|
|
print(f"#####DEBUG: {new_client_doc.doctype}:", new_client_doc.as_dict())
|
|
|
|
# Handle address creation
|
|
print("#####DEBUG: Checking for existing address for customer/lead:", 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:
|
|
return build_error_response("Address already exists for this customer.", 400)
|
|
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")
|
|
}).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")})
|
|
print("Contact exists check:", contact_exists)
|
|
if not contact_exists:
|
|
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":1 if 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
|
|
}]
|
|
}).insert(ignore_permissions=True)
|
|
print("Created new contact:", contact_doc.as_dict())
|
|
else:
|
|
contact_doc = frappe.get_doc("Contact", {"email_id": contact_data.get("email"), "phone": contact_data.get("phone_number")})
|
|
print("Contact already exists:", contact_doc.as_dict())
|
|
contact_docs.append(contact_doc)
|
|
|
|
##### Create links
|
|
# Customer -> Address
|
|
if new_client_doc.doctype == "Customer":
|
|
print("#####DEBUG: Creating links between customer, address, and contacts.")
|
|
new_client_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:", new_client_doc.as_dict())
|
|
new_client_doc.append("custom_add_contacts", {
|
|
"contact": contact_doc.name,
|
|
"email": contact_doc.custom_email,
|
|
"phone": contact_doc.phone,
|
|
"role": contact_doc.role
|
|
})
|
|
new_client_doc.save(ignore_permissions=True)
|
|
|
|
# Address -> Customer/Lead
|
|
print("#####DEBUG: Linking address to customer.")
|
|
address_doc.append("links", {
|
|
"link_doctype": new_client_doc.doctype,
|
|
"link_name": new_client_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
|
|
})
|
|
|
|
address_doc.save(ignore_permissions=True)
|
|
|
|
# Contact -> Customer/Lead & Address
|
|
print("#####DEBUG: Linking contacts to customer.")
|
|
for contact_doc in contact_docs:
|
|
contact_doc.address = address_doc.name
|
|
contact_doc.append("links", {
|
|
"link_doctype": new_client_doc.doctype,
|
|
"link_name": new_client_doc.name
|
|
})
|
|
contact_doc.append("links", {
|
|
"link_doctype": "Address",
|
|
"link_name": address_doc.name
|
|
})
|
|
contact_doc.custom_customer = new_client_doc.name
|
|
contact_doc.save(ignore_permissions=True)
|
|
|
|
frappe.local.message_log = []
|
|
return build_success_response({
|
|
"customer": new_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",
|
|
pluck="name")
|
|
return build_success_response(client_names)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|