306 lines
No EOL
14 KiB
Python
306 lines
No EOL
14 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."""
|
|
try:
|
|
clientData = {"addresses": []}
|
|
customer = frappe.get_doc("Customer", client_name)
|
|
clientData = {**clientData, **customer.as_dict()}
|
|
addresses = frappe.db.get_all("Address", fields=["*"], filters={"custom_customer_to_bill": client_name})
|
|
for address in addresses if addresses else []:
|
|
addressData = {"jobs": []}
|
|
addressData = {**addressData, **address}
|
|
addressData["estimates"] = frappe.db.get_all("Quotation", fields=["*"], filters={"custom_installation_address": address.address_title})
|
|
addressData["onsite_meetings"] = frappe.db.get_all("On-Site Meeting", fields=["*"], filters={"address": address.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(addressData)
|
|
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)
|
|
|
|
|
|
# address = frappe.get_doc("Address", client_name)
|
|
# customer_name = address.custom_customer_to_bill if address.custom_customer_to_bill else [link.link_name for link in address.links if link.link_doctype == "Customer"][0] if address.links else None
|
|
# if not customer_name:
|
|
# raise Exception(f"No customer linked to address {client_name}. Suggested fix: Ensure the address is linked to a customer via the ERPnext UI.")
|
|
# project_names = frappe.db.get_all("Project", fields=["name"], or_filters=[
|
|
# ["custom_installation_address", "=", address.address_title],
|
|
# ["custom_address", "=", address.address_title]
|
|
# ], limit_page_length=100)
|
|
# # contacts = [] # currently not needed as the customer doctype comes with contacts
|
|
# onsite_meetings = frappe.db.get_all(
|
|
# "On-Site Meeting",
|
|
# fields=["*"],
|
|
# filters={"address": address.address_title}
|
|
# )
|
|
# quotations = frappe.db.get_all(
|
|
# "Quotation",
|
|
# fields=["*"],
|
|
# filters={"custom_installation_address": address.address_title}
|
|
# )
|
|
# sales_orders = []
|
|
# projects = [frappe.get_doc("Project", proj["name"]) for proj in project_names]
|
|
# sales_invoices = []
|
|
# payment_entries = frappe.db.get_all(
|
|
# "Payment Entry",
|
|
# fields=["*"],
|
|
# filters={"party_type": "Customer"},
|
|
# or_filters=[
|
|
# ["party", "=", customer_name],
|
|
# ["party_name", "=", customer_name]
|
|
# ])
|
|
# payment_orders = []
|
|
# jobs = []
|
|
# for project in projects:
|
|
# job = []
|
|
# jobs.append(job)
|
|
# customer = frappe.get_doc("Customer", customer_name)
|
|
# # get all associated data as needed
|
|
# return build_success_response({
|
|
# "address": address,
|
|
# "customer": customer,
|
|
# # "contacts": [], # currently not needed as the customer doctype comes with contacts
|
|
# "jobs": jobs,
|
|
# "sales_invoices": sales_invoices,
|
|
# "payment_entries": payment_entries,
|
|
# "sales_orders": sales_orders,
|
|
# "quotations": quotations,
|
|
# "onsite_meetings": onsite_meetings,
|
|
# })
|
|
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
|
|
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", customer)
|
|
|
|
print("Customer:", customer_doc.as_dict())
|
|
|
|
# Check for existing address
|
|
filters = {
|
|
"address_title": data.get("address_title"),
|
|
}
|
|
existing_address = frappe.db.exists("Address", filters)
|
|
print("Existing address check:", existing_address)
|
|
if existing_address:
|
|
frappe.throw(f"Address already exists for customer {data.get('customer_name')}.", frappe.ValidationError)
|
|
|
|
# Create address
|
|
address_doc = frappe.get_doc({
|
|
"doctype": "Address",
|
|
"address_line1": data.get("address_line1"),
|
|
"city": data.get("city"),
|
|
"state": data.get("state"),
|
|
"country": "United States",
|
|
"address_title": data.get("address_title"),
|
|
"pincode": data.get("pincode"),
|
|
"custom_customer_to_bill": customer_doc.name
|
|
}).insert(ignore_permissions=True)
|
|
|
|
# Link address to customer
|
|
link = {
|
|
"link_doctype": "Customer",
|
|
"link_name": customer_doc.name
|
|
}
|
|
address_doc.append("links", link)
|
|
address_doc.save(ignore_permissions=True)
|
|
|
|
return build_success_response({
|
|
"customer": customer_doc.as_dict(),
|
|
"address": address_doc.as_dict()
|
|
})
|
|
except frappe.ValidationError as ve:
|
|
return build_error_response(str(ve), 400)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500) |