223 lines
No EOL
11 KiB
Python
223 lines
No EOL
11 KiB
Python
import frappe
|
|
from frappe.model.document import Document
|
|
import requests
|
|
from .contact_service import ContactService, DbService
|
|
|
|
class AddressService:
|
|
|
|
@staticmethod
|
|
def build_address_title(customer_name, address_data) -> str:
|
|
"""Build a title for the address based on its fields."""
|
|
print(f"DEBUG: Building address title for customer '{customer_name}' with address data: {address_data}")
|
|
is_billing = address_data.get("is_billing_address", False)
|
|
address_type = "Billing" if is_billing else "Service"
|
|
return f"{customer_name} - {address_data.get('address_line1', '')} {address_data.get('city')} - {address_type}"
|
|
|
|
@staticmethod
|
|
def build_full_dict(
|
|
address_doc: Document,
|
|
included_links: list = ["contacts", "on-site meetings", "quotations", "sales orders", "projects", "companies"]) -> frappe._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) -> Document:
|
|
"""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 = DbService.get_or_throw("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."""
|
|
print(f"DEBUG: Checking existence of Address with name: {address_name}")
|
|
result = frappe.db.exists("Address", address_name) is not None
|
|
print(f"DEBUG: Address existence: {result}")
|
|
return result
|
|
|
|
@staticmethod
|
|
def get(address_name: str) -> Document:
|
|
"""Retrieve an address document by name. Returns None if not found."""
|
|
print(f"DEBUG: Retrieving Address document with name: {address_name}")
|
|
if AddressService.exists(address_name):
|
|
address_doc = DbService.get_or_throw("Address", address_name)
|
|
print("DEBUG: Address document found.")
|
|
return address_doc
|
|
print("DEBUG: Address document not found.")
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_or_throw(address_name: str) -> Document:
|
|
"""Retrieve an address document by name or throw an error if not found."""
|
|
address_doc = AddressService.get(address_name)
|
|
if address_doc:
|
|
return address_doc
|
|
raise ValueError(f"Address with name {address_name} does not exist.")
|
|
|
|
@staticmethod
|
|
def update_value(doc_name: str, fieldname: str, value, save: bool = True) -> Document:
|
|
"""Update a specific field value of a document."""
|
|
print(f"DEBUG: Updating Address {doc_name}, setting {fieldname} to {value}")
|
|
if AddressService.exists(doc_name) is False:
|
|
raise ValueError(f"Address with name {doc_name} does not exist.")
|
|
if save:
|
|
print("DEBUG: Saving updated Address document.")
|
|
address_doc = AddressService.get_or_throw(doc_name)
|
|
setattr(address_doc, fieldname, value)
|
|
address_doc.save(ignore_permissions=True)
|
|
else:
|
|
print("DEBUG: Not saving Address document as save=False.")
|
|
frappe.db.set_value("Address", doc_name, fieldname, value)
|
|
print(f"DEBUG: Updated Address {doc_name}: set {fieldname} to {value}")
|
|
return address_doc
|
|
|
|
@staticmethod
|
|
def get_value(doc_name: str, fieldname: str) -> any:
|
|
"""Get a specific field value of a document. Returns None if document does not exist."""
|
|
print(f"DEBUG: Getting value of field {fieldname} from Address {doc_name}")
|
|
if not AddressService.exists(doc_name):
|
|
print("DEBUG: Value cannot be retrieved; Address does not exist.")
|
|
return None
|
|
value = frappe.db.get_value("Address", doc_name, fieldname)
|
|
print(f"DEBUG: Retrieved value: {value}")
|
|
return value
|
|
|
|
@staticmethod
|
|
def get_value_or_throw(doc_name: str, fieldname: str) -> any:
|
|
"""Get a specific field value of a document or throw an error if document does not exist."""
|
|
value = AddressService.get_value(doc_name, fieldname)
|
|
if value is not None:
|
|
return value
|
|
raise ValueError(f"Address with name {doc_name} does not exist.")
|
|
|
|
@staticmethod
|
|
def create(address_data: dict) -> Document:
|
|
"""Create a new address."""
|
|
print("DEBUG: Creating new Address with data:", address_data)
|
|
address = frappe.get_doc({
|
|
"doctype": "Address",
|
|
**address_data
|
|
})
|
|
address.insert(ignore_permissions=True)
|
|
print("DEBUG: Created new Address:", address.as_dict())
|
|
return address
|
|
|
|
@staticmethod
|
|
def link_address_to_customer(address_doc: Document, customer_type: str, customer_name: str):
|
|
"""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.append("links", {
|
|
"link_doctype": customer_type,
|
|
"link_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: Document, contact_name: str):
|
|
"""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.append("links", {
|
|
"link_doctype": "Contact",
|
|
"link_name": 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: dict) -> Document:
|
|
"""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)
|
|
print("DEBUG: Appending link:", link)
|
|
address_doc.append(field, link)
|
|
print("DEBUG: Saving address document after appending link.")
|
|
address_doc.save(ignore_permissions=True)
|
|
print(f"DEBUG: Set link field {field} for Address {address_name} with link data {link}")
|
|
|
|
@staticmethod
|
|
def get_county_and_set(address_doc: Document, save: bool = False):
|
|
"""Get the county from the address document and set it if not already set."""
|
|
if not address_doc.county:
|
|
print(f"DEBUG: Getting county for Address {address_doc.name}")
|
|
# Example logic to determine county from address fields
|
|
# This is a placeholder; actual implementation may vary
|
|
url = "https://geocoding.geo.cencus.gov/geocoder/geographies/coordinates"
|
|
params = {
|
|
"x": address_doc.longitude,
|
|
"y": address_doc.latitude,
|
|
"benchmark": "Public_AR_Current",
|
|
"vintage": "Current_Current",
|
|
"format": "json"
|
|
}
|
|
|
|
r = requests.get(url, params=params, timeout=10)
|
|
data = r.json()
|
|
|
|
try:
|
|
county = data['result']['geographies']['Counties'][0]['NAME']
|
|
county_fips = data['result']['geographies']['Counties'][0]['GEOID']
|
|
except (KeyError, IndexError):
|
|
return None
|
|
|
|
county_info = {
|
|
"county": county,
|
|
"county_fips": county_fips
|
|
}
|
|
|
|
AddressService.update_value(address_doc.name, "county", county_info, save)
|
|
AddressService.update_value(address_doc.name, "county_fips", county_fips, save)
|
|
|