125 lines
No EOL
5 KiB
Python
125 lines
No EOL
5 KiB
Python
import frappe
|
|
from custom_ui.db_utils import build_error_response, build_success_response
|
|
|
|
@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)
|
|
except Exception as e:
|
|
return build_error_response(str(e), 500)
|
|
|
|
@frappe.whitelist()
|
|
def get_address(address_name):
|
|
"""Get a specific address by name."""
|
|
try:
|
|
address = frappe.get_doc("Address", 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()
|
|
def get_addresses(fields=["*"], filters={}):
|
|
"""Get addresses with optional filtering."""
|
|
if isinstance(fields, str):
|
|
import json
|
|
fields = json.loads(fields)
|
|
if isinstance(filters, str):
|
|
import json
|
|
filters = json.loads(filters)
|
|
if fields[0] != "*" and len(fields) == 1:
|
|
pluck = fields[0]
|
|
fields = None
|
|
print(f"Getting addresses with fields: {fields} and filters: {filters} and pluck: {pluck}")
|
|
try:
|
|
addresses = frappe.get_all(
|
|
"Address",
|
|
fields=fields,
|
|
filters=filters,
|
|
order_by="address_line1 desc",
|
|
pluck=pluck
|
|
)
|
|
return build_success_response(addresses)
|
|
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 address_exists(address_line1, address_line2, city, state, pincode):
|
|
"""Check if an address with the given details already exists."""
|
|
filters = {
|
|
"address_line1": address_line1,
|
|
"address_line2": address_line2,
|
|
"city": city,
|
|
"state": state,
|
|
"pincode": pincode
|
|
}
|
|
return frappe.db.exists("Address", filters) 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])
|
|
address_doc.append("links", {
|
|
"link_doctype": client_doc.doctype,
|
|
"link_name": client_doc.name
|
|
})
|
|
setattr(address_doc, "custom_customer_to_bill" if client_doc.doctype == "Customer" else "lead_name", client_doc.name)
|
|
# Address -> Contact
|
|
print("#####DEBUG: Linking contacts to address.")
|
|
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.append("links", {
|
|
"link_doctype": "Contact",
|
|
"link_name": contact_doc.name
|
|
})
|
|
address_doc.save(ignore_permissions=True) |