427 lines
No EOL
19 KiB
Python
427 lines
No EOL
19 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
|
|
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":
|
|
clientData["addresses"] = customer.custom_select_address or []
|
|
clientData["contacts"] = customer.custom_add_contacts or []
|
|
|
|
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())
|
|
doctypes_to_fetch_history = []
|
|
# 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
|
|
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
|
|
client_doc = check_and_get_client_doc(customer_name)
|
|
if not client_doc:
|
|
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")
|
|
})
|
|
|
|
#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.") |