bnilsen/custom_ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

big update

Browse source
This commit is contained in:
Casey 2026-01-15 08:36:08 -06:00
parent d53ebf9ecd
commit 5c7e93fcc7
26 changed files with 1890 additions and 423 deletions
Download patch file Download diff file Expand all files Collapse all files

69
custom_ui/api/db/addresses.py
View file

@ -1,31 +1,15 @@
import frappe
import json
from custom_ui.db_utils import build_error_response, build_success_response
from custom_ui.services import ClientService, AddressService
@frappe.whitelist()
def get_address_by_full_address(full_address):
"""Get address by full_address, including associated contacts."""
print(f"DEBUG: get_address_by_full_address called with full_address: {full_address}")
try:
address = frappe.get_doc("Address", {"full_address": full_address}).as_dict()
customer_exists = frappe.db.exists("Customer", address.get("custom_customer_to_bill"))
doctype = "Customer" if customer_exists else "Lead"
name = ""
if doctype == "Customer":
name = address.get("custom_customer_to_bill")
else:
## filter through links for one with doctype Lead
lead_links = address.get("links", [])
print(f"DEBUG: lead_links: {lead_links}")
lead_name = [link.link_name for link in lead_links if link.link_doctype == "Lead"]
name = lead_name[0] if lead_name else ""
address["customer"] = frappe.get_doc(doctype, name).as_dict()
contacts = []
for contact_link in address.custom_linked_contacts:
contact_doc = frappe.get_doc("Contact", contact_link.contact)
contacts.append(contact_doc.as_dict())
address["contacts"] = contacts
return build_success_response(address)
address = AddressService.get_address_by_full_address(full_address)
return build_success_response(AddressService.build_full_dict(address))
except Exception as e:
return build_error_response(str(e), 500)
@ -33,23 +17,23 @@ def get_address_by_full_address(full_address):
def get_address(address_name):
"""Get a specific address by name."""
try:
address = frappe.get_doc("Address", address_name)
address = AddressService.get_or_throw(address_name)
return build_success_response(address.as_dict())
except Exception as e:
return build_error_response(str(e), 500)
@frappe.whitelist()
def get_contacts_for_address(address_name):
"""Get contacts linked to a specific address."""
try:
address = frappe.get_doc("Address", address_name)
contacts = []
for contact_link in address.custom_linked_contacts:
contact = frappe.get_doc("Contact", contact_link.contact)
contacts.append(contact.as_dict())
return build_success_response(contacts)
except Exception as e:
return build_error_response(str(e), 500)
# @frappe.whitelist() #### DEPRECATED FUNCTION
# def get_contacts_for_address(address_name):
# """Get contacts linked to a specific address."""
# try:
# address = AddressService.get_or_throw(address_name)
# contacts = []
# for contact_link in address.custom_linked_contacts:
# contact = frappe.get_doc("Contact", contact_link.contact)
# contacts.append(contact.as_dict())
# return build_success_response(contacts)
# except Exception as e:
# return build_error_response(str(e), 500)
@frappe.whitelist()
def get_addresses(fields=["*"], filters={}):
@ -74,16 +58,6 @@ def get_addresses(fields=["*"], filters={}):
except Exception as e:
frappe.log_error(message=str(e), title="Get Addresses Failed")
return build_error_response(str(e), 500)
def create_address(address_data):
"""Create a new address."""
address = frappe.get_doc({
"doctype": "Address",
**address_data
})
address.insert(ignore_permissions=True)
return address
def update_address(address_data):
"""Update an existing address."""
@ -106,19 +80,10 @@ def address_exists(address_line1, address_line2, city, state, pincode):
}
return frappe.db.exists("Address", filters) is not None
def check_and_get_address_by_name(address_name):
"""Check if an address exists by name and return the address document if found."""
if frappe.db.exists("Address", address_name):
return frappe.get_doc("Address", address_name)
raise ValueError(f"Address with name {address_name} does not exist.")
def address_exists_by_name(address_name):
"""Check if an address with the given name exists."""
return frappe.db.exists("Address", address_name) is not None
def calculate_address_title(customer_name, address_data):
return f"{customer_name} - {address_data.get('address_line1', '')}, {address_data.get('city', '')} - {address_data.get('type', '')}"
def create_address_links(address_doc, client_doc, contact_docs):
print("#####DEBUG: Linking customer to address.")
print("#####DEBUG: Client Doc:", client_doc.as_dict(), "Address Doc:", address_doc.as_dict(), "Contact Docs:", [c.as_dict() for c in contact_docs])

55
custom_ui/api/db/bid_meetings.py
View file

@ -1,7 +1,7 @@
import frappe
import json
from custom_ui.db_utils import build_error_response, build_success_response, process_filters, process_sorting
from custom_ui.services import DbService
from custom_ui.services import DbService, ClientService, AddressService
@frappe.whitelist()
def get_week_bid_meetings(week_start, week_end):
@ -67,38 +67,51 @@ def get_unscheduled_bid_meetings():
@frappe.whitelist()
def create_bid_meeting(address, notes="", company=None, contact=None):
"""Create a new On-Site Meeting with Unscheduled status."""
def get_bid_meeting(name):
"""Get a specific On-Site Meeting by name."""
try:
print(f"DEBUG: Creating meeting with address='{address}', notes='{notes}', company='{company}'")
# Validate address parameter
if not address or address == "None" or not address.strip():
return build_error_response("Address is required and cannot be empty.", 400)
# Get the address document name from the full address string
address_name = frappe.db.get_value("Address", filters={"full_address": address}, fieldname="name")
meeting = frappe.get_doc("On-Site Meeting", name)
meeting_dict = meeting.as_dict()
print(f"DEBUG: Address lookup result: address_name='{address_name}'")
# Get the full address data
if meeting_dict.get("address"):
address_doc = frappe.get_doc("Address", meeting_dict["address"])
meeting_dict["address"] = address_doc.as_dict()
if not address_name:
return build_error_response(f"Address '{address}' not found in the system.", 404)
address_doc = DbService.get("Address", address_name)
return build_success_response(meeting_dict)
except frappe.DoesNotExistError:
return build_error_response(f"On-Site Meeting '{name}' does not exist.", 404)
except Exception as e:
frappe.log_error(message=str(e), title="Get On-Site Meeting Failed")
return build_error_response(str(e), 500)
@frappe.whitelist()
def create_bid_meeting(data):
"""Create a new On-Site Meeting with Unscheduled status."""
if isinstance(data, str):
data = json.loads(data)
try:
print(f"DEBUG: Creating meeting with data='{data}'")
address_doc = DbService.get_or_throw("Address", data.get("address"))
# Create the meeting with Unscheduled status
meeting = frappe.get_doc({
"doctype": "On-Site Meeting",
"address": address_doc.name,
"notes": notes or "",
"notes": data.get("notes") or "",
"status": "Unscheduled",
"company": company,
"contact": contact,
"company": data.get("company"),
"contact": data.get("contact"),
"party_type": address_doc.customer_type,
"party_name": address_doc.customer_name
"party_name": address_doc.customer_name,
"project_template": data.get("project_template")
})
meeting.flags.ignore_permissions = True
meeting.insert(ignore_permissions=True)
# ClientService.append_link(address_doc.customer_name, "onsite_meetings", "onsite_meeting", meeting.name)
# AddressService.append_link(address_doc.name, "onsite_meetings", "onsite_meeting", meeting.name)
meeting.flags.ignore_permissions = True
frappe.db.commit()
# Clear any auto-generated messages from Frappe

156
custom_ui/api/db/clients.py
View file

@ -1,8 +1,9 @@
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.addresses import address_exists
from custom_ui.api.db.contacts import check_and_get_contact, create_contact, create_contact_links
from custom_ui.services import AddressService, ContactService, ClientService
# ===============================================================================
# CLIENT MANAGEMENT API METHODS
@ -96,7 +97,7 @@ 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": []}
clientData = {"addresses": [], "contacts": [], "jobs": [], "sales_invoices": [], "payment_entries": [], "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)
@ -142,6 +143,30 @@ def get_client(client_name):
return build_error_response(str(ve), 400)
except Exception as e:
return build_error_response(str(e), 500)
@frappe.whitelist()
def get_client_v2(client_name):
"""Get detailed information for a specific client including address, customer, and projects."""
print("DEBUG: get_client_v2 called with client_name:", client_name)
try:
clientData = {"addresses": [], "jobs": [], "payment_entries": [], "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()}
clientData["contacts"] = [ContactService.get_or_throw(link.contact) for link in clientData["contacts"]]
clientData["addresses"] = [AddressService.get_or_throw(link.address) for link in clientData["properties"]]
if clientData["doctype"] == "Lead":
clientData["customer_name"] = customer.custom_customer_name
# 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()
@ -189,8 +214,7 @@ def get_clients_table_data(filters={}, sortings=[], page=1, page_size=10):
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"
customer_name = frappe.get_value("Lead", address.get("customer_name"), "custom_customer_name")
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
@ -202,9 +226,9 @@ def get_clients_table_data(filters={}, sortings=[], page=1, page_size=10):
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["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)
@ -256,19 +280,21 @@ def upsert_client(data):
customer_name = data.get("customer_name")
contacts = data.get("contacts", [])
addresses = data.get("addresses", [])
# 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)
for address in addresses:
if address_exists(
address.get("address_line1"),
address.get("address_line2"),
address.get("city"),
address.get("state"),
address.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
@ -280,28 +306,16 @@ def upsert_client(data):
"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
"customer_type": customer_type,
"companies": [{ "company": data.get("company_name")
}]
}
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:
@ -316,12 +330,14 @@ def upsert_client(data):
)
if not contact_doc:
print("#####DEBUG: No existing contact found. Creating new contact.")
contact_doc = create_contact({
contact_doc = ContactService.create({
"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,
"customer_type": "Lead",
"customer_name": client_doc.name,
"email_ids": [{
"email_id": contact_data.get("email"),
"is_primary": 1
@ -332,37 +348,61 @@ def upsert_client(data):
"is_primary_phone": 1
}]
})
ContactService.link_contact_to_customer(contact_doc, "Lead", client_doc.name)
contact_docs.append(contact_doc)
# Link all contacts to client after creating them
client_doc.reload()
for idx, contact_data in enumerate(contacts):
if isinstance(contact_data, str):
contact_data = json.loads(contact_data)
contact_doc = contact_docs[idx]
client_doc.append("contacts", {
"contact": contact_doc.name
})
if contact_data.get("is_primary", False):
client_doc.primary_contact = contact_doc.name
client_doc.save(ignore_permissions=True)
# ##### 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)
# Handle address creation
address_docs = []
for address in addresses:
print("#####DEBUG: Creating address with data:", address)
address_doc = AddressService.create_address({
"address_title": build_address_title(customer_name, address),
"address_line1": address.get("address_line1"),
"address_line2": address.get("address_line2"),
"city": address.get("city"),
"state": address.get("state"),
"country": "United States",
"pincode": address.get("pincode"),
"customer_type": "Lead",
"customer_name": client_doc.name,
"companies": [{ "company": data.get("company_name") }]
})
AddressService.link_address_to_customer(address_doc, "Lead", client_doc.name)
address_doc.reload()
for contact_to_link_idx in address.get("contacts", []):
contact_doc = contact_docs[contact_to_link_idx]
AddressService.link_address_to_contact(address_doc, contact_doc.name)
address_doc.reload()
ContactService.link_contact_to_address(contact_doc, address_doc.name)
primary_contact = contact_docs[address.get("primary_contact)", 0)]
AddressService.set_primary_contact(address_doc.name, primary_contact.name)
address_docs.append(address_doc)
# Link all addresses to client after creating them
client_doc.reload()
for address_doc in address_docs:
client_doc.append("properties", {
"address": address_doc.name
})
client_doc.save(ignore_permissions=True)
frappe.local.message_log = []
return build_success_response({
"customer": client_doc.as_dict(),
"address": address_doc.as_dict(),
"address": [address_doc.as_dict() for address_doc in address_docs],
"contacts": [contact_doc.as_dict() for contact_doc in contact_docs]
})
except frappe.ValidationError as ve:

41
custom_ui/api/db/estimates.py
View file

@ -4,6 +4,7 @@ from custom_ui.api.db.general import get_doc_history
from custom_ui.db_utils import process_query_conditions, build_datatable_dict, get_count_or_filters, build_success_response, build_error_response
from werkzeug.wrappers import Response
from custom_ui.api.db.clients import check_if_customer, convert_lead_to_customer
from custom_ui.services import DbService, ClientService, AddressService
# ===============================================================================
# ESTIMATES & INVOICES API METHODS
@ -244,11 +245,11 @@ def update_response(name, response):
if accepted:
template = "custom_ui/templates/estimates/accepted.html"
if check_if_customer(estimate.party_name):
print("DEBUG: Party is already a customer:", estimate.party_name)
else:
print("DEBUG: Converting lead to customer for party:", estimate.party_name)
convert_lead_to_customer(estimate.party_name)
# if check_if_customer(estimate.party_name):
# print("DEBUG: Party is already a customer:", estimate.party_name)
# else:
# print("DEBUG: Converting lead to customer for party:", estimate.party_name)
# convert_lead_to_customer(estimate.party_name)
elif response == "Requested call":
template = "custom_ui/templates/estimates/request-call.html"
else:
@ -384,7 +385,8 @@ def upsert_estimate(data):
print("DEBUG: Upsert estimate data:", data)
estimate_name = data.get("estimate_name")
is_customer = True if frappe.db.exists("Customer", data.get("customer")) else False
client_doctype = ClientService.get_client_doctype(data.get("customer"))
project_template = data.get("project_template", None)
# If estimate_name exists, update existing estimate
if estimate_name:
@ -392,11 +394,21 @@ def upsert_estimate(data):
estimate = frappe.get_doc("Quotation", estimate_name)
# Update fields
estimate.custom_installation_address = data.get("address")
estimate.party_name = data.get("customer")
estimate.contact_person = data.get("contact_name")
# estimate.custom_installation_address = data.get("address")
# estimate.custom_job_address = data.get("address_name")
# estimate.party_name = data.get("customer")
# estimate.contact_person = data.get("contact_name")
estimate.custom_requires_half_payment = data.get("requires_half_payment", 0)
estimate.custom_project_template = project_template
estimate.custom_quotation_template = data.get("quotation_template", None)
# estimate.company = data.get("company")
# estimate.contact_email = data.get("contact_email")
# estimate.quotation_to = client_doctype
# estimate.customer_name = data.get("customer")
# estimate.customer_address = data.get("address_name")
# estimate.letter_head = data.get("company")
# estimate.from_onsite_meeting = data.get("onsite_meeting", None)
# Clear existing items and add new ones
estimate.items = []
for item in data.get("items", []):
@ -418,6 +430,7 @@ def upsert_estimate(data):
else:
print("DEBUG: Creating new estimate")
print("DEBUG: Retrieved address name:", data.get("address_name"))
client_doctype = ClientService.get_client_doctype(data.get("customer"))
new_estimate = frappe.get_doc({
"doctype": "Quotation",
"custom_requires_half_payment": data.get("requires_half_payment", 0),
@ -425,13 +438,15 @@ def upsert_estimate(data):
"custom_current_status": "Draft",
"contact_email": data.get("contact_email"),
"party_name": data.get("customer"),
"quotation_to": "Customer" if is_customer else "Lead",
"quotation_to": client_doctype,
"company": data.get("company"),
"customer_name": data.get("customer"),
"customer": data.get("customer"),
"customer_type": client_doctype,
"customer_address": data.get("address_name"),
"contact_person": data.get("contact_name"),
"letter_head": data.get("company"),
"custom_project_template": data.get("project_template", None),
"custom_quotation_template": data.get("quotation_template", None),
"from_onsite_meeting": data.get("onsite_meeting", None)
})
for item in data.get("items", []):
@ -443,6 +458,8 @@ def upsert_estimate(data):
"discount_percentage": item.get("discount_percentage") or item.get("discountPercentage", 0)
})
new_estimate.insert()
AddressService.append_link(data.get("address_name"), "quotations", "quotation", new_estimate.name)
ClientService.append_link(data.get("customer"), "quotations", "quotation", new_estimate.name)
print("DEBUG: New estimate created with name:", new_estimate.name)
return build_success_response(new_estimate.as_dict())
except Exception as e:

13
custom_ui/api/db/jobs.py
View file

@ -201,3 +201,16 @@ def get_install_projects(start_date=None, end_date=None):
return {"status": "success", "data": calendar_events}
except Exception as e:
return {"status": "error", "message": str(e)}
@frappe.whitelist()
def get_project_templates_for_company(company_name):
"""Get project templates for a specific company."""
try:
templates = frappe.get_all(
"Project Template",
fields=["*"],
filters={"company": company_name}
)
return build_success_response(templates)
except Exception as e:
return build_error_response(str(e), 500),

5
custom_ui/events/address.py
View file

@ -1,9 +1,8 @@
import frappe
from custom_ui.db_utils import build_full_address
def after_insert(doc, method):
print(doc.as_dict())
def before_insert(doc, method):
print("DEBUG: Before Insert Triggered for Address")
if not doc.full_address:
doc.full_address = build_full_address(doc)
doc.save()

112
custom_ui/events/estimate.py
View file

@ -1,67 +1,69 @@
import frappe
from erpnext.selling.doctype.quotation.quotation import make_sales_order
from custom_ui.services import DbService
from custom_ui.services import DbService, AddressService, ClientService
def after_insert(doc, method):
print("DEBUG: after_insert hook triggered for Quotation:", doc.name)
try:
template = doc.custom_project_template or "Other"
if template == "Other":
print("WARN: No project template specified.")
if template == "SNW Install":
print("DEBUG: SNW Install template detected, updating custom address field.")
DbService.set_value(
doctype="Address",
name=doc.custom_job_address,
fieldname="custom_estimate_sent_status",
value="Pending"
)
except Exception as e:
print("ERROR in after_insert hook:", str(e))
frappe.log_error(f"Error in estimate after_insert: {str(e)}", "Estimate Hook Error")
print("DEBUG: After insert hook triggered for Quotation:", doc.name)
AddressService.append_link_v2(
doc.custom_job_address,
{"quotations": {"quotation": doc.name, "project_template": doc.custom_project_template}}
)
template = doc.custom_project_template or "Other"
if template == "Other":
print("WARN: No project template specified.")
if template == "SNW Install":
print("DEBUG: SNW Install template detected, updating custom address field.")
AddressService.update_value(
doc.custom_job_address,
"estimate_sent_status",
"In Progress"
)
def before_insert(doc, method):
print("DEBUG: Before insert hook triggered for Quotation:", doc.name)
# if doc.custom_project_template == "SNW Install":
# print("DEBUG: Quotation uses SNW Install template, setting initial Address status to 'In Progress'.")
# address_doc = AddressService.get_or_throw(doc.custom_job_address)
# if "SNW Install" in [link.project_template for link in address_doc.quotations]:
# raise frappe.ValidationError("An Estimate with project template 'SNW Install' is already linked to this address.")
def after_save(doc, method):
print("DEBUG: after_save hook triggered for Quotation:", doc.name)
if doc.custom_sent and doc.custom_response and doc.custom_project_template == "SNW Install":
print("DEBUG: Quotation has been sent, updating Address status")
try:
DbService.set_value(
doctype="Address",
name=doc.custom_job_address,
fieldname="custom_estimate_sent_status",
value="Sent"
def before_submit(doc, method):
print("DEBUG: Before submit hook triggered for Quotation:", doc.name)
if doc.custom_project_template == "SNW Install":
print("DEBUG: Quotation uses SNW Install template.")
if doc.custom_current_status == "Estimate Sent":
print("DEBUG: Current status is 'Estimate Sent', updating Address status to 'Sent'.")
AddressService.update_value(
doc.custom_job_address,
"estimate_sent_status",
"Completed"
)
except Exception as e:
print("ERROR updating Address in after_save hook:", str(e))
frappe.log_error(f"Error updating Address in estimate after_save: {str(e)}", "Estimate Hook Error")
def on_update_after_submit(doc, method):
print("DEBUG: on_update_after_submit hook triggered for Quotation:", doc.name)
print("DEBUG: Current custom_current_status:", doc.custom_current_status)
if doc.custom_current_status == "Estimate Accepted":
doc.custom_current_status = "Won"
if doc.custom_project_template == "SNW Install":
DbService.set_value(
doctype="Address",
name=doc.custom_job_address,
fieldname="custom_estimate_sent_status",
value="Completed"
)
try:
print("DEBUG: Creating Sales Order from accepted Estimate")
new_sales_order = make_sales_order(doc.name)
new_sales_order.custom_requires_half_payment = doc.requires_half_payment
new_sales_order.custom_installation_address = doc.custom_installation_address
new_sales_order.payment_schedule = []
print("DEBUG: Setting payment schedule for Sales Order")
new_sales_order.set_payment_schedule()
print("DEBUG: Inserting Sales Order:", new_sales_order.as_dict())
new_sales_order.delivery_date = new_sales_order.transaction_date
new_sales_order.insert()
print("DEBUG: Submitting Sales Order")
new_sales_order.submit()
frappe.db.commit()
print("DEBUG: Sales Order created successfully:", new_sales_order.name)
except Exception as e:
print("ERROR creating Sales Order from Estimate:", str(e))
frappe.log_error(f"Error creating Sales Order from Estimate {doc.name}: {str(e)}", "Estimate on_update_after_submit Error")
print("DEBUG: Quotation marked as Won, updating current status.")
if doc.customer_type == "Lead":
print("DEBUG: Customer is a Lead, converting to Customer and updating Quotation.")
new_customer = ClientService.convert_lead_to_customer(doc.customer, update_quotations=False)
doc.customer = new_customer.name
doc.customer_type = "Customer"
doc.save()
print("DEBUG: Creating Sales Order from accepted Estimate")
new_sales_order = make_sales_order(doc.name)
new_sales_order.custom_requires_half_payment = doc.requires_half_payment
new_sales_order.customer = doc.customer
# new_sales_order.custom_installation_address = doc.custom_installation_address
# new_sales_order.custom_job_address = doc.custom_job_address
new_sales_order.payment_schedule = []
print("DEBUG: Setting payment schedule for Sales Order")
new_sales_order.set_payment_schedule()
print("DEBUG: Inserting Sales Order:", new_sales_order.as_dict())
new_sales_order.delivery_date = new_sales_order.transaction_date
new_sales_order.insert()
print("DEBUG: Submitting Sales Order")
new_sales_order.submit()
frappe.db.commit()
print("DEBUG: Sales Order created successfully:", new_sales_order.name)

47
custom_ui/events/onsite_meeting.py
View file

@ -1,21 +1,40 @@
import frappe
from custom_ui.services import DbService, AddressService, ClientService
def before_insert(doc, method):
print("DEBUG: Before Insert Triggered for On-Site Meeting")
if doc.project_template == "SNW Install":
address_doc = AddressService.get_or_throw(doc.address)
# Address.onsite_meetings is a child table with two fields: onsite_meeting (Link) and project_template (Link). Iterate through to see if there is already an SNW Install meeting linked.
for link in address_doc.onsite_meetings:
if link.project_template == "SNW Install":
raise frappe.ValidationError("An On-Site Meeting with project template 'SNW Install' is already linked to this address.")
def after_insert(doc, method):
print("DEBUG: After Insert Triggered for On-Site Meeting")
print("DEBUG: Updating on-site meeting status in Address")
if doc.address and not doc.end_time and not doc.start_time:
address_doc = frappe.get_doc("Address", doc.address)
address_doc.custom_onsite_meeting_scheduled = "In Progress"
address_doc.save()
print("DEBUG: Linking bid meeting to customer and address")
AddressService.append_link_v2(doc.address, "onsite_meetings", {"onsite_meeting": doc.name, "project_template": doc.project_template})
ClientService.append_link(doc.party_name, "onsite_meetings", "onsite_meeting", doc.name)
if doc.project_template == "SNW Install":
print("DEBUG: Project template is SNW Install, updating Address status to In Progress")
AddressService.update_value(doc.address, "onsite_meeting_scheduled", "In Progress")
def after_save(doc, method):
print("DEBUG: After Save Triggered for On-Site Meeting")
if doc.status == "Completed":
print("DEBUG: Meeting marked as Completed, updating Address status")
address_doc = frappe.get_doc("Address", doc.address)
address_doc.custom_onsite_meeting_scheduled = "Completed"
address_doc.save()
return
def before_save(doc, method):
print("DEBUG: Before Save Triggered for On-Site Meeting")
if doc.status != "Scheduled" and doc.start_time and doc.end_time:
print("DEBUG: Meeting has start and end time, setting status to Scheduled")
doc.status = "Scheduled"
doc.save()
if doc.project_template == "SNW Install":
print("DEBUG: Project template is SNW Install")
if doc.status == "Completed":
print("DEBUG: Meeting marked as Completed, updating Address status")
current_status = AddressService.get_value(doc.address, "onsite_meeting_scheduled")
if current_status != doc.status:
AddressService.update_value(doc.address, "onsite_meeting_scheduled", "Completed")
def validate_address_link(doc, method):
print("DEBUG: Validating Address link for On-Site Meeting")
if doc.onsite_meeting:
meeting = DbService.get_or_throw("On-Site Meeting", doc.onsite_meeting)
doc.project_template = meeting.project_template

33
custom_ui/hooks.py
View file

@ -161,19 +161,21 @@ add_to_apps_screen = [
doc_events = {
"On-Site Meeting": {
"after_insert": "custom_ui.events.onsite_meeting.after_insert",
"on_update": "custom_ui.events.onsite_meeting.after_save"
"before_save": "custom_ui.events.onsite_meeting.before_save",
"before_insert": "custom_ui.events.onsite_meeting.before_insert"
},
"Address": {
"after_insert": "custom_ui.events.address.after_insert"
"before_insert": "custom_ui.events.address.before_insert"
},
"Quotation": {
"before_insert": "custom_ui.events.estimate.before_insert",
"after_insert": "custom_ui.events.estimate.after_insert",
"on_update": "custom_ui.events.estimate.after_save",
"after_submit": "custom_ui.events.estimate.after_submit",
# "before_save": "custom_ui.events.estimate.before_save",
"before_submit": "custom_ui.events.estimate.before_submit",
"on_update_after_submit": "custom_ui.events.estimate.on_update_after_submit"
},
"Sales Order": {
"on_submit": "custom_ui.events.sales_order.on_submit",
"on_submit": "custom_ui.events.sales_order.on_submit"
},
"Task": {
"before_insert": "custom_ui.events.task.before_insert"
@ -189,7 +191,25 @@ fixtures = [
"Quotation Template Item",
"Customer Company Link",
"Customer Address Link",
"Customer Contact Link"
"Customer Contact Link",
# New link doctypes
"Customer Project Link",
"Customer Quotation Link",
"Customer Sales Order Link",
"Customer On-Site Meeting Link",
"Lead Address Link",
"Lead Contact Link",
"Lead Companies Link",
"Lead Quotation Link",
"Lead On-Site Meeting Link",
"Address Project Link",
"Address Quotation Link",
"Address On-Site Meeting Link",
"Address Sales Order Link",
"Address Contact Link",
"Address Company Link",
"Contact Address Link",
]]
]
},
@ -227,6 +247,7 @@ fixtures = [
]
# Scheduled Tasks
# ---------------

7
custom_ui/services/__init__.py
View file

@ -1,3 +1,6 @@
from .address_service import AddressService
from .db_service import DbService
from .contact_service import ContactService
from .db_service import DbService
from .client_service import ClientService
from .estimate_service import EstimateService
from .onsite_meeting_service import OnSiteMeetingService

92
custom_ui/services/address_service.py
View file

@ -1,7 +1,42 @@
import frappe
from .contact_service import ContactService, DbService
class AddressService:
@staticmethod
def build_full_dict(
address_doc,
included_links: list = ["contacts", "on-site meetings", "quotations", "sales orders", "projects", "companies"]) -> dict:
"""Build a full dictionary representation of an address, including all links. Can optionally exclude links."""
print(f"DEBUG: Building full dict for Address {address_doc.name}")
address_dict = address_doc.as_dict()
if "contacts" in included_links:
address_dict["contacts"] = [ContactService.get_or_throw(link.contact).as_dict() for link in address_doc.contacts]
if "on-site meetings" in included_links:
address_dict["onsite_meetings"] = [DbService.get_or_throw("On-Site Meeting", link.onsite_meeting).as_dict() for link in address_doc.onsite_meetings]
if "quotations" in included_links:
address_dict["quotations"] = [DbService.get_or_throw("Quotation", link.quotation).as_dict() for link in address_doc.quotations]
if "sales orders" in included_links:
address_dict["sales_orders"] = [DbService.get_or_throw("Sales Order", link.sales_order).as_dict() for link in address_doc.sales_orders]
if "projects" in included_links:
address_dict["projects"] = [DbService.get_or_throw("Project", link.project).as_dict() for link in address_doc.projects]
if "companies" in included_links:
address_dict["companies"] = [DbService.get_or_throw("Company", link.company).as_dict() for link in address_doc.companies]
print(f"DEBUG: Built full dict for Address {address_doc.name}: {address_dict}")
return address_dict
@staticmethod
def get_address_by_full_address(full_address: str):
"""Retrieve an address document by its full_address field. Returns None if not found."""
print(f"DEBUG: Retrieving Address document with full_address: {full_address}")
address_name = frappe.db.get_value("Address", {"full_address": full_address})
if address_name:
address_doc = frappe.get_doc("Address", address_name)
print("DEBUG: Address document found.")
return address_doc
print("DEBUG: Address document not found.")
return None
@staticmethod
def exists(address_name: str) -> bool:
"""Check if an address with the given name exists."""
@ -76,4 +111,59 @@ class AddressService:
address.insert(ignore_permissions=True)
print("DEBUG: Created new Address:", address.as_dict())
return address
@staticmethod
def link_address_to_customer(address_doc, customer_type, customer_name):
"""Link an address to a customer or lead."""
print(f"DEBUG: Linking Address {address_doc.name} to {customer_type} {customer_name}")
address_doc.customer_type = customer_type
address_doc.customer_name = customer_name
address_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Address {address_doc.name} to {customer_type} {customer_name}")
@staticmethod
def link_address_to_contact(address_doc, contact_name):
"""Link an address to a contact."""
print(f"DEBUG: Linking Address {address_doc.name} to Contact {contact_name}")
address_doc.append("contacts", {
"contact": contact_name
})
address_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Address {address_doc.name} to Contact {contact_name}")
@staticmethod
def create_address(address_data):
"""Create a new address."""
address = frappe.get_doc({
"doctype": "Address",
**address_data
})
address.insert(ignore_permissions=True)
return address
@staticmethod
def set_primary_contact(address_name: str, contact_name: str):
"""Set the primary contact for an address."""
print(f"DEBUG: Setting primary contact for Address {address_name} to Contact {contact_name}")
frappe.db.set_value("Address", address_name, "primary_contact", contact_name)
print(f"DEBUG: Set primary contact for Address {address_name} to Contact {contact_name}")
@staticmethod
def append_link(address_name: str, field: str, link_doctype: str, link_name: str):
"""Set a link field for an address."""
print(f"DEBUG: Setting link field {field} for Address {address_name} to {link_doctype} {link_name}")
address_doc = AddressService.get_or_throw(address_name)
address_doc.append(field, {
link_doctype.lower(): link_name
})
address_doc.save(ignore_permissions=True)
print(f"DEBUG: Set link field {field} for Address {address_name} to {link_doctype} {link_name}")
@staticmethod
def append_link_v2(address_name: str, field: str, link: dict):
"""Set a link field for an address using a link dictionary."""
print(f"DEBUG: Setting link field {field} for Address {address_name} with link data {link}")
address_doc = AddressService.get_or_throw(address_name)
address_doc.append(field, link)
address_doc.save(ignore_permissions=True)
print(f"DEBUG: Set link field {field} for Address {address_name} with link data {link}")

71
custom_ui/services/client_service.py Normal file
View file

@ -0,0 +1,71 @@
import frappe
from .db_service import DbService
from erpnext.crm.doctype.lead.lead import make_customer
from .address_service import AddressService
from .contact_service import ContactService
from .estimate_service import EstimateService
from .onsite_meeting_service import OnSiteMeetingService
class ClientService:
@staticmethod
def get_client_doctype(client_name: str) -> str:
"""Determine if the client is a Customer or Lead."""
if DbService.exists("Customer", client_name):
return "Customer"
elif DbService.exists("Lead", client_name):
return "Lead"
else:
raise ValueError(f"Client with name {client_name} does not exist as Customer or Lead.")
@staticmethod
def set_primary_contact(client_name: str, contact_name: str):
"""Set the primary contact for a client (Customer or Lead)."""
print(f"DEBUG: Setting primary contact for client {client_name} to contact {contact_name}")
client_doctype = ClientService.get_client_doctype(client_name)
frappe.db.set_value(client_doctype, client_name, "primary_contact", contact_name)
print(f"DEBUG: Set primary contact for client {client_name} to contact {contact_name}")
@staticmethod
def append_link(client_name: str, field: str, link_doctype: str, link_name: str):
"""Set a link field for a client (Customer or Lead)."""
print(f"DEBUG: Setting link field {field} for client {client_name} to {link_doctype} {link_name}")
client_doctype = ClientService.get_client_doctype(client_name)
client_doc = frappe.get_doc(client_doctype, client_name)
client_doc.append(field, {
link_doctype.lower(): link_name
})
client_doc.save(ignore_permissions=True)
print(f"DEBUG: Set link field {field} for client {client_doc.get('name')} to {link_doctype} {link_name}")
@staticmethod
def convert_lead_to_customer(
lead_name: str,
update_quotations: bool = True,
update_addresses: bool = True,
update_contacts: bool = True,
update_onsite_meetings: bool = True
):
"""Convert a Lead to a Customer."""
print(f"DEBUG: Converting Lead {lead_name} to Customer")
lead_doc = DbService.get_or_throw("Lead", lead_name)
customer_doc = make_customer(lead_doc.name)
customer_doc.insert(ignore_permissions=True)
if update_addresses:
for address in lead_doc.get("addresses", []):
address_doc = AddressService.get_or_throw(address.get("address"))
AddressService.link_address_to_customer(address_doc, "Customer", customer_doc.name)
if update_contacts:
for contact in lead_doc.get("contacts", []):
contact_doc = ContactService.get_or_throw(contact.get("contact"))
ContactService.link_contact_to_customer(contact_doc, "Customer", customer_doc.name)
if update_quotations:
for quotation in lead_doc.get("quotations", []):
quotation_doc = EstimateService.get_or_throw(quotation.get("quotation"))
EstimateService.link_estimate_to_customer(quotation_doc, "Customer", customer_doc.name)
if update_onsite_meetings:
for meeting in lead_doc.get("onsite_meetings", []):
meeting_doc = OnSiteMeetingService.get_or_throw(meeting.get("onsite_meeting"))
OnSiteMeetingService.link_onsite_meeting_to_customer(meeting_doc, "Customer", customer_doc.name)
print(f"DEBUG: Converted Lead {lead_name} to Customer {customer_doc.name}")
return customer_doc

40
custom_ui/services/contact_service.py Normal file
View file

@ -0,0 +1,40 @@
import frappe
from .db_service import DbService
class ContactService:
@staticmethod
def create(data: dict):
"""Create a new contact."""
print("DEBUG: Creating new Contact with data:", data)
contact = frappe.get_doc({
"doctype": "Contact",
**data
})
contact.insert(ignore_permissions=True)
print("DEBUG: Created new Contact:", contact.as_dict())
return contact
@staticmethod
def link_contact_to_customer(contact_doc, customer_type, customer_name):
"""Link a contact to a customer or lead."""
print(f"DEBUG: Linking Contact {contact_doc.name} to {customer_type} {customer_name}")
contact_doc.customer_type = customer_type
contact_doc.customer_name = customer_name
contact_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Contact {contact_doc.name} to {customer_type} {customer_name}")
@staticmethod
def link_contact_to_address(contact_doc, address_name):
"""Link an address to a contact."""
print(f"DEBUG: Linking Address {address_name} to Contact {contact_doc.name}")
contact_doc.append("addresses", {
"address": address_name
})
contact_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Address {address_name} to Contact {contact_doc.name}")
@staticmethod
def get_or_throw(contact_name: str):
"""Retrieve a Contact document or throw an error if it does not exist."""
return DbService.get_or_throw("Contact", contact_name)

9
custom_ui/services/estimate_service.py
View file

@ -84,4 +84,13 @@ class EstimateService:
estimate_doc.insert(ignore_permissions=True)
print(f"DEBUG: Created Quotation document: {estimate_doc.as_dict()}")
return estimate_doc
@staticmethod
def link_estimate_to_customer(estimate_doc: frappe._dict, customer_type: str, customer_name: str) -> None:
"""Link a Quotation document to a client document."""
print(f"DEBUG: Linking Quotation {estimate_doc.name} to {customer_type} {customer_name}")
estimate_doc.customer_type = customer_type
estimate_doc.customer = customer_name
estimate_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Quotation {estimate_doc.name} to {customer_type} {customer_name}")

38
custom_ui/services/onsite_meeting_service.py Normal file
View file

@ -0,0 +1,38 @@
import frappe
class OnSiteMeetingService:
@staticmethod
def exists(onsite_meeting_name: str) -> bool:
"""Check if an OnSite Meeting document exists by name."""
result = frappe.db.exists("OnSite Meeting", onsite_meeting_name) is not None
print(f"DEBUG: OnSite Meeting existence for {onsite_meeting_name}: {result}")
return result
@staticmethod
def get(onsite_meeting_name: str) -> frappe._dict:
"""Retrieve an OnSite Meeting document by name. Returns None if not found."""
print(f"DEBUG: Retrieving OnSite Meeting document with name: {onsite_meeting_name}")
if OnSiteMeetingService.exists(onsite_meeting_name):
onsite_meeting_doc = frappe.get_doc("OnSite Meeting", onsite_meeting_name)
print("DEBUG: OnSite Meeting document found.")
return onsite_meeting_doc
print("DEBUG: OnSite Meeting document not found.")
return None
@staticmethod
def get_or_throw(onsite_meeting_name: str) -> frappe._dict:
"""Retrieve an OnSite Meeting document or throw an error if not found."""
onsite_meeting_doc = OnSiteMeetingService.get(onsite_meeting_name)
if not onsite_meeting_doc:
raise ValueError(f"OnSite Meeting with name {onsite_meeting_name} does not exist.")
return onsite_meeting_doc
@staticmethod
def link_onsite_meeting_to_customer(onsite_meeting_doc, customer_type, customer_name):
"""Link an onsite meeting to a customer or lead."""
print(f"DEBUG: Linking Onsite Meeting {onsite_meeting_doc.name} to {customer_type} {customer_name}")
onsite_meeting_doc.party_type = customer_type
onsite_meeting_doc.party_name = customer_name
onsite_meeting_doc.save(ignore_permissions=True)
print(f"DEBUG: Linked Onsite Meeting {onsite_meeting_doc.name} to {customer_type} {customer_name}")