CaWittMN08/custom_ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

clean up db method files

Browse Source
This commit is contained in:
Casey Wittrock 2025-11-12 10:25:41 -06:00
parent a675af5cf8
commit b087ea673e
6 changed files with 199 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
custom_ui/api/db/__init__.py Normal file
View File

160
custom_ui/api/db.py → custom_ui/api/db/clients.py
View File

@ -1,9 +1,13 @@
import frappe, json, re
from datetime import datetime, date
import frappe, json
from custom_ui.db_utils import process_query_conditions, build_datatable_response, get_count_or_filters
# ===============================================================================
# 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
base_filters = {}
if weekly and week_start_date and week_end_date:
@ -78,8 +82,10 @@ def get_client_status_counts(weekly=False, week_start_date=None, week_end_date=N
return categories
@frappe.whitelist()
def get_client(client_name):
"""Get detailed information for a specific client including address, customer, and projects."""
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
project_names = frappe.db.get_all("Project", fields=["name"], or_filters=[
@ -100,20 +106,24 @@ def get_client(client_name):
@frappe.whitelist()
def get_clients_table_data(filters={}, sortings=[], page=1, page_size=10):
"""Get paginated client table data with filtering and sorting support."""
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)
count = frappe.db.count(
"Address",
filters=processed_filters
) if not is_or else frappe.db.sql(
*get_count_or_filters("Address", processed_filters)
)[0][0]
# 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"],
@ -123,6 +133,7 @@ def get_clients_table_data(filters = {}, sortings = [], page=1, page_size=10):
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:
@ -151,30 +162,12 @@ def get_clients_table_data(filters = {}, sortings = [], page=1, page_size=10):
return build_datatable_response(data=tableRows, count=count, page=page, page_size=page_size)
@frappe.whitelist()
def upsert_estimate(data):
pass
@frappe.whitelist()
def upsert_job(data):
pass
@frappe.whitelist()
def upsert_invoice(data):
pass
@frappe.whitelist()
def upsert_warranty(data):
pass
@frappe.whitelist()
def upsert_client(data):
"""Create or update a client (customer and address)."""
data = json.loads(data)
"""
Upsert a document in the database.
If a document with the same name exists, it will be updated.
Otherwise, a new document will be created.
"""
# Handle customer creation/update
customer = frappe.db.exists("Customer", {"customer_name": data.get("customer_name")})
if not customer:
customer_doc = frappe.get_doc({
@ -184,7 +177,10 @@ def upsert_client(data):
}).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"),
}
@ -192,6 +188,8 @@ def upsert_client(data):
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"),
@ -202,6 +200,8 @@ def upsert_client(data):
"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
@ -214,105 +214,3 @@ def upsert_client(data):
"address": address_doc,
"success": True
}
@frappe.whitelist()
def get_jobs_table_data(filters = {}, sortings = [], page=1, page_size=10):
print("DEBUG: Raw job options received:", filters, sortings, page, page_size)
processed_filters, processed_sortings, is_or, page, page_size = process_query_conditions(filters, sortings, page, page_size)
count = frappe.db.count(
"Project",
filters=processed_filters if not is_or else None,
or_filters=processed_filters if is_or else None,
) if not is_or else frappe.db.sql(
build
)
projects = frappe.db.get_all(
"Project",
fields=["*"],
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
)
tableRows = []
for project in projects:
tableRow = {}
tableRow["id"] = project["name"]
tableRow["name"] = project["name"]
tableRow["installation_address"] = project.get("custom_installation_address", "")
tableRow["customer"] = project.get("customer", "")
tableRow["status"] = project.get("status", "")
tableRow["percent_complete"] = project.get("percent_complete", 0)
tableRows.append(tableRow)
return build_datatable_response(data=tableRows, count=count, page=page, page_size=page_size)
@frappe.whitelist()
def get_warranty_claims(filters = {}, sortings = [], page=1, page_size=10):
print("DEBUG: Raw warranty options received:", filters, sortings, page, page_size)
processed_filters, processed_sortings, is_or, page, page_size = process_query_conditions(filters, sortings, page, page_size)
count = frappe.db.count(
"Warranty Claim",
filters=processed_filters if not is_or else None,
or_filters=processed_filters if is_or else None
)
warranty_claims = frappe.db.get_all(
"Warranty Claim",
fields=["*"],
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
)
tableRows = []
for warranty in warranty_claims:
tableRow = {}
tableRow["id"] = warranty["name"]
tableRow["warrantyId"] = warranty["name"]
tableRow["customer"] = warranty.get("customer_name", "")
tableRow["serviceAddress"] = warranty.get("service_address", warranty.get("address_display", ""))
# Extract a brief description from the complaint HTML
complaint_text = warranty.get("complaint", "")
if complaint_text:
# Simple HTML stripping for display - take first 100 chars
clean_text = re.sub('<.*?>', '', complaint_text)
clean_text = clean_text.strip()
if len(clean_text) > 100:
clean_text = clean_text[:100] + "..."
tableRow["issueDescription"] = clean_text
else:
tableRow["issueDescription"] = ""
tableRow["status"] = warranty.get("status", "")
tableRow["complaintDate"] = warranty.get("complaint_date", "")
tableRow["complaintRaisedBy"] = warranty.get("complaint_raised_by", "")
tableRow["fromCompany"] = warranty.get("from_company", "")
tableRow["territory"] = warranty.get("territory", "")
tableRow["resolutionDate"] = warranty.get("resolution_date", "")
tableRow["warrantyStatus"] = warranty.get("warranty_amc_status", "")
# Add priority based on status and date (can be customized)
if warranty.get("status") == "Open":
# Calculate priority based on complaint date
if warranty.get("complaint_date"):
complaint_date = warranty.get("complaint_date")
if isinstance(complaint_date, str):
complaint_date = datetime.strptime(complaint_date, "%Y-%m-%d").date()
elif isinstance(complaint_date, datetime):
complaint_date = complaint_date.date()
days_old = (date.today() - complaint_date).days
if days_old > 7:
tableRow["priority"] = "High"
elif days_old > 3:
tableRow["priority"] = "Medium"
else:
tableRow["priority"] = "Low"
else:
tableRow["priority"] = "Medium"
else:
tableRow["priority"] = "Low"
tableRows.append(tableRow)
return build_datatable_response(data=tableRows, count=count, page=page, page_size=page_size)

19
custom_ui/api/db/estimates.py Normal file
View File

@ -0,0 +1,19 @@
import frappe, json
from custom_ui.db_utils import process_query_conditions, build_datatable_response, get_count_or_filters
# ===============================================================================
# ESTIMATES & INVOICES API METHODS
# ===============================================================================
@frappe.whitelist()
def upsert_estimate(data):
"""Create or update an estimate."""
# TODO: Implement estimate creation/update logic
pass
@frappe.whitelist()
def upsert_invoice(data):
"""Create or update an invoice."""
# TODO: Implement invoice creation/update logic
pass

49
custom_ui/api/db/jobs.py Normal file
View File

@ -0,0 +1,49 @@
import frappe, json
from custom_ui.db_utils import process_query_conditions, build_datatable_response, get_count_or_filters
# ===============================================================================
# JOB MANAGEMENT API METHODS
# ===============================================================================
@frappe.whitelist()
def get_jobs_table_data(filters={}, sortings=[], page=1, page_size=10):
"""Get paginated job table data with filtering and sorting support."""
print("DEBUG: Raw job options received:", filters, sortings, page, page_size)
processed_filters, processed_sortings, is_or, page, page_size = process_query_conditions(filters, sortings, page, page_size)
# Handle count with proper OR filter support
if is_or:
count = frappe.db.sql(*get_count_or_filters("Project", processed_filters))[0][0]
else:
count = frappe.db.count("Project", filters=processed_filters)
projects = frappe.db.get_all(
"Project",
fields=["*"],
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
)
tableRows = []
for project in projects:
tableRow = {}
tableRow["id"] = project["name"]
tableRow["name"] = project["name"]
tableRow["installation_address"] = project.get("custom_installation_address", "")
tableRow["customer"] = project.get("customer", "")
tableRow["status"] = project.get("status", "")
tableRow["percent_complete"] = project.get("percent_complete", 0)
tableRows.append(tableRow)
return build_datatable_response(data=tableRows, count=count, page=page, page_size=page_size)
@frappe.whitelist()
def upsert_job(data):
"""Create or update a job (project)."""
# TODO: Implement job creation/update logic
pass

90
custom_ui/api/db/warranties.py Normal file
View File

@ -0,0 +1,90 @@
import frappe, json, re
from datetime import datetime, date
from custom_ui.db_utils import process_query_conditions, build_datatable_response, get_count_or_filters
# ===============================================================================
# WARRANTY MANAGEMENT API METHODS
# ===============================================================================
@frappe.whitelist()
def get_warranty_claims(filters={}, sortings=[], page=1, page_size=10):
"""Get paginated warranty claims table data with filtering and sorting support."""
print("DEBUG: Raw warranty options received:", filters, sortings, page, page_size)
processed_filters, processed_sortings, is_or, page, page_size = process_query_conditions(filters, sortings, page, page_size)
# Handle count with proper OR filter support
if is_or:
count = frappe.db.sql(*get_count_or_filters("Warranty Claim", processed_filters))[0][0]
else:
count = frappe.db.count("Warranty Claim", filters=processed_filters)
warranty_claims = frappe.db.get_all(
"Warranty Claim",
fields=["*"],
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
)
tableRows = []
for warranty in warranty_claims:
tableRow = {}
tableRow["id"] = warranty["name"]
tableRow["warrantyId"] = warranty["name"]
tableRow["customer"] = warranty.get("customer_name", "")
tableRow["serviceAddress"] = warranty.get("service_address", warranty.get("address_display", ""))
# Extract a brief description from the complaint HTML
complaint_text = warranty.get("complaint", "")
if complaint_text:
# Simple HTML stripping for display - take first 100 chars
clean_text = re.sub('<.*?>', '', complaint_text)
clean_text = clean_text.strip()
if len(clean_text) > 100:
clean_text = clean_text[:100] + "..."
tableRow["issueDescription"] = clean_text
else:
tableRow["issueDescription"] = ""
tableRow["status"] = warranty.get("status", "")
tableRow["complaintDate"] = warranty.get("complaint_date", "")
tableRow["complaintRaisedBy"] = warranty.get("complaint_raised_by", "")
tableRow["fromCompany"] = warranty.get("from_company", "")
tableRow["territory"] = warranty.get("territory", "")
tableRow["resolutionDate"] = warranty.get("resolution_date", "")
tableRow["warrantyStatus"] = warranty.get("warranty_amc_status", "")
# Add priority based on status and date (can be customized)
if warranty.get("status") == "Open":
# Calculate priority based on complaint date
if warranty.get("complaint_date"):
complaint_date = warranty.get("complaint_date")
if isinstance(complaint_date, str):
complaint_date = datetime.strptime(complaint_date, "%Y-%m-%d").date()
elif isinstance(complaint_date, datetime):
complaint_date = complaint_date.date()
days_old = (date.today() - complaint_date).days
if days_old > 7:
tableRow["priority"] = "High"
elif days_old > 3:
tableRow["priority"] = "Medium"
else:
tableRow["priority"] = "Low"
else:
tableRow["priority"] = "Medium"
else:
tableRow["priority"] = "Low"
tableRows.append(tableRow)
return build_datatable_response(data=tableRows, count=count, page=page, page_size=page_size)
@frappe.whitelist()
def upsert_warranty(data):
"""Create or update a warranty claim."""
# TODO: Implement warranty creation/update logic
pass

14
frontend/src/api.js
View File

@ -2,13 +2,13 @@ import DataUtils from "./utils";
const ZIPPOPOTAMUS_BASE_URL = "https://api.zippopotam.us/us";
const FRAPPE_PROXY_METHOD = "custom_ui.api.proxy.request";
const FRAPPE_UPSERT_CLIENT_METHOD = "custom_ui.api.db.upsert_client";
const FRAPPE_UPSERT_ESTIMATE_METHOD = "custom_ui.api.db.upsert_estimate";
const FRAPPE_UPSERT_JOB_METHOD = "custom_ui.api.db.upsert_job";
const FRAPPE_UPSERT_INVOICE_METHOD = "custom_ui.api.db.upsert_invoice";
const FRAPPE_GET_CLIENT_STATUS_COUNTS_METHOD = "custom_ui.api.db.get_client_status_counts";
const FRAPPE_GET_CLIENT_TABLE_DATA_METHOD = "custom_ui.api.db.get_clients_table_data";
const FRAPPE_GET_CLIENT_METHOD = "custom_ui.api.db.get_client";
const FRAPPE_UPSERT_ESTIMATE_METHOD = "custom_ui.api.db.estimates.upsert_estimate";
const FRAPPE_UPSERT_JOB_METHOD = "custom_ui.api.db.jobs.upsert_job";
const FRAPPE_UPSERT_INVOICE_METHOD = "custom_ui.api.db.invoices.upsert_invoice";
const FRAPPE_UPSERT_CLIENT_METHOD = "custom_ui.api.db.clients.upsert_client";
const FRAPPE_GET_CLIENT_STATUS_COUNTS_METHOD = "custom_ui.api.db.clients.get_client_status_counts";
const FRAPPE_GET_CLIENT_TABLE_DATA_METHOD = "custom_ui.api.db.clients.get_clients_table_data";
const FRAPPE_GET_CLIENT_METHOD = "custom_ui.api.db.clients.get_client";
class Api {
static async request(frappeMethod, args = {}) {