1031 lines
35 KiB
Python
1031 lines
35 KiB
Python
import frappe
|
|
from frappe import _
|
|
from frappe.utils import nowdate
|
|
import json
|
|
|
|
|
|
def _ok(payload, code=200):
|
|
frappe.response.status_code = code
|
|
frappe.response.message = payload
|
|
|
|
|
|
def _err(msg, code=500):
|
|
frappe.response.status_code = code
|
|
frappe.response.message = {"error": msg}
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_number_cards():
|
|
"""
|
|
Returns counts for Number Cards:
|
|
- total_assets
|
|
- work_orders_open
|
|
- work_orders_in_progress
|
|
- work_orders_completed
|
|
"""
|
|
try:
|
|
total_assets = frappe.db.count(
|
|
"Asset",
|
|
{"custom_delete_status": ["!=", "Deleted"]},
|
|
)
|
|
work_orders_open = frappe.db.count("Work Order", {"status": ["in", ["Not Started", "Open", "Pending"]]})
|
|
work_orders_in_progress = frappe.db.count("Work Order", {"status": ["in", ["In Process", "In Progress", "Started"]]})
|
|
work_orders_completed = frappe.db.count("Work Order", {"status": ["in", ["Completed", "Closed", "Finished"]]})
|
|
|
|
_ok({
|
|
"total_assets": total_assets,
|
|
"work_orders_open": work_orders_open,
|
|
"work_orders_in_progress": work_orders_in_progress,
|
|
"work_orders_completed": work_orders_completed,
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_number_cards")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def list_dashboard_charts(search=None, public_only=True, limit=50):
|
|
"""
|
|
List available Dashboard Chart docs and their y-axis rows.
|
|
"""
|
|
try:
|
|
filters = {}
|
|
if str(public_only) in ("1", "true", "True"): # tolerate string flags
|
|
filters["is_public"] = 1
|
|
|
|
charts = frappe.get_all(
|
|
"Dashboard Chart",
|
|
filters=filters,
|
|
fields=[
|
|
"name",
|
|
"chart_name",
|
|
"type",
|
|
"is_public",
|
|
"chart_type",
|
|
"report_name",
|
|
"use_report_chart",
|
|
"x_field",
|
|
"time_interval",
|
|
"timespan",
|
|
"custom_options",
|
|
],
|
|
limit=int(limit or 50),
|
|
order_by="modified desc",
|
|
)
|
|
|
|
for c in charts:
|
|
y_rows = frappe.get_all(
|
|
"Dashboard Chart Field",
|
|
filters={"parenttype": "Dashboard Chart", "parent": c["name"]},
|
|
fields=["y_field", "color"],
|
|
order_by="idx asc",
|
|
)
|
|
c["y_axes"] = y_rows
|
|
|
|
_ok({"charts": charts})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "list_dashboard_charts")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_dashboard_chart_data(chart_name, report_filters=None):
|
|
"""
|
|
Return chart-ready JSON for any Dashboard Chart (Report-based or Custom).
|
|
Supports:
|
|
- Report-based charts (with or without use_report_chart)
|
|
- Custom/Document-type based charts
|
|
- Multi-series (stacked) bar charts
|
|
"""
|
|
try:
|
|
if isinstance(report_filters, str):
|
|
report_filters = json.loads(report_filters or "{}")
|
|
report_filters = report_filters or {}
|
|
|
|
# Try to find chart by name first, then by chart_name
|
|
chart = None
|
|
if frappe.db.exists("Dashboard Chart", chart_name):
|
|
chart = frappe.get_doc("Dashboard Chart", chart_name)
|
|
else:
|
|
# Search by chart_name field
|
|
chart_doc_name = frappe.db.get_value("Dashboard Chart", {"chart_name": chart_name}, "name")
|
|
if chart_doc_name:
|
|
chart = frappe.get_doc("Dashboard Chart", chart_doc_name)
|
|
else:
|
|
# Try case-insensitive search
|
|
chart_doc_name = frappe.db.sql("""
|
|
SELECT name FROM `tabDashboard Chart`
|
|
WHERE LOWER(chart_name) = LOWER(%s) OR LOWER(name) = LOWER(%s)
|
|
LIMIT 1
|
|
""", (chart_name, chart_name), as_dict=True)
|
|
if chart_doc_name:
|
|
chart = frappe.get_doc("Dashboard Chart", chart_doc_name[0].name)
|
|
|
|
if not chart:
|
|
_err(f"Dashboard Chart '{chart_name}' not found. Use list_dashboard_charts API to see available charts.")
|
|
return
|
|
|
|
# Get y-axes configuration
|
|
y_axes = frappe.get_all(
|
|
"Dashboard Chart Field",
|
|
filters={"parenttype": "Dashboard Chart", "parent": chart.name},
|
|
fields=["y_field", "color"],
|
|
order_by="idx asc",
|
|
)
|
|
|
|
# PRIORITY 1: Handle Report-based charts (check report_name first)
|
|
if chart.report_name:
|
|
return _process_report_chart(chart, y_axes, report_filters)
|
|
|
|
# PRIORITY 2: Handle Custom/Document-type charts (only if NO report_name)
|
|
if chart.chart_type == "Custom" or (chart.document_type and chart.based_on):
|
|
return _process_custom_chart(chart, y_axes)
|
|
|
|
# PRIORITY 3: Handle Group By / Count / Sum charts
|
|
if chart.chart_type in ("Group By", "Count", "Sum"):
|
|
return _process_group_by_chart(chart)
|
|
|
|
_err(f"Unsupported chart type: {chart.chart_type} for chart: {chart.name}")
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_dashboard_chart_data")
|
|
_err(str(e))
|
|
|
|
|
|
def _process_report_chart(chart, y_axes, report_filters):
|
|
"""
|
|
Process a Report-based Dashboard Chart.
|
|
Runs the report and extracts data based on x_field and y_axes configuration.
|
|
"""
|
|
try:
|
|
# Merge chart's default filters with provided filters
|
|
chart_filters = _parse_custom_options(chart.filters_json) if chart.filters_json else {}
|
|
merged_filters = {**chart_filters, **report_filters}
|
|
|
|
# Run the report
|
|
run = frappe.get_attr("frappe.desk.query_report.run")
|
|
report_result = run(chart.report_name, filters=merged_filters)
|
|
|
|
# Report charts that ship their own chart payload (use_report_chart)
|
|
if chart.use_report_chart and report_result.get("chart"):
|
|
report_chart = report_result["chart"]
|
|
chart_data = report_chart.get("data") or {}
|
|
labels = [str(l) for l in (chart_data.get("labels") or [])]
|
|
raw_datasets = chart_data.get("datasets") or []
|
|
datasets = []
|
|
for ds in raw_datasets:
|
|
values = []
|
|
for val in ds.get("values") or []:
|
|
try:
|
|
values.append(float(val) if val is not None else 0)
|
|
except (ValueError, TypeError):
|
|
values.append(0)
|
|
datasets.append({
|
|
"name": ds.get("name") or "value",
|
|
"values": values,
|
|
})
|
|
if not datasets:
|
|
datasets = [{"name": "value", "values": []}]
|
|
|
|
chart_type = (report_chart.get("type") or chart.type or "Bar").title()
|
|
custom_options = _parse_custom_options(chart.custom_options)
|
|
report_options = report_chart.get("custom_options")
|
|
if isinstance(report_options, dict):
|
|
custom_options = {**custom_options, **report_options}
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": datasets,
|
|
"type": chart_type,
|
|
"options": custom_options,
|
|
"source": {"report": chart.report_name},
|
|
})
|
|
return
|
|
|
|
rows = report_result.get("result", []) or []
|
|
|
|
# Filter out total rows
|
|
data_rows = [r for r in rows if not (isinstance(r, dict) and r.get("is_total_row"))]
|
|
|
|
# Get x-axis labels
|
|
x_key = chart.x_field
|
|
labels = []
|
|
for r in data_rows:
|
|
if isinstance(r, dict):
|
|
val = r.get(x_key)
|
|
if val is not None:
|
|
labels.append(str(val))
|
|
|
|
# Build datasets from y-axes
|
|
datasets = []
|
|
|
|
# Color palette for multi-series charts
|
|
default_colors = [
|
|
"#6366F1", # Indigo
|
|
"#10B981", # Green
|
|
"#3B82F6", # Blue
|
|
"#F59E0B", # Amber
|
|
"#EC4899", # Pink
|
|
"#8B5CF6", # Purple
|
|
"#06B6D4", # Cyan
|
|
"#EF4444", # Red
|
|
]
|
|
|
|
for idx, y in enumerate(y_axes):
|
|
y_field = y.get("y_field")
|
|
series_name = y_field # Use field name as series name
|
|
values = []
|
|
|
|
for r in data_rows:
|
|
if isinstance(r, dict):
|
|
val = r.get(y_field)
|
|
try:
|
|
values.append(float(val) if val is not None else 0)
|
|
except (ValueError, TypeError):
|
|
values.append(0)
|
|
|
|
# Use provided color or default from palette
|
|
color = y.get("color") or default_colors[idx % len(default_colors)]
|
|
datasets.append({
|
|
"name": series_name,
|
|
"values": values,
|
|
"color": color
|
|
})
|
|
|
|
chart_type = (chart.type or "Bar").title()
|
|
custom_options = _parse_custom_options(chart.custom_options)
|
|
|
|
# Handle Pie charts (only use first dataset)
|
|
if chart_type.lower() == "pie":
|
|
ds = datasets[0] if datasets else {"name": "value", "values": []}
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [ds],
|
|
"type": "Pie",
|
|
"options": custom_options,
|
|
"source": {"report": chart.report_name},
|
|
})
|
|
return
|
|
|
|
# Return data for Bar/Line charts (supports multi-series/stacked)
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": datasets,
|
|
"type": chart_type,
|
|
"options": custom_options,
|
|
"source": {"report": chart.report_name},
|
|
# Include result for frontend transformation if needed
|
|
"result": data_rows if len(y_axes) > 1 else None,
|
|
})
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), f"Report Chart Error: {chart.name}")
|
|
_err(f"Error processing report chart: {str(e)}")
|
|
|
|
|
|
def _process_custom_chart(chart, y_axes):
|
|
"""
|
|
Process a Custom/Document-type based Dashboard Chart.
|
|
Queries the source doctype directly based on configuration.
|
|
"""
|
|
try:
|
|
source = chart.document_type
|
|
based_on = chart.based_on
|
|
value_based_on = chart.value_based_on or "name"
|
|
|
|
if not source or not based_on:
|
|
_err(f"Custom chart '{chart.name}' missing document_type or based_on configuration")
|
|
return
|
|
|
|
# Build aggregation query
|
|
if chart.type and chart.type.lower() == "pie":
|
|
# Pie chart: Group by based_on field and count
|
|
data = frappe.db.sql(f"""
|
|
SELECT `{based_on}` as label, COUNT(`{value_based_on}`) as value
|
|
FROM `tab{source}`
|
|
GROUP BY `{based_on}`
|
|
ORDER BY value DESC
|
|
""", as_dict=True)
|
|
|
|
labels = [str(d.get("label") or "Unknown") for d in data]
|
|
values = [float(d.get("value") or 0) for d in data]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": "count", "values": values}],
|
|
"type": "Pie",
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"doctype": source},
|
|
})
|
|
else:
|
|
# Bar chart: Group by based_on
|
|
data = frappe.db.sql(f"""
|
|
SELECT `{based_on}` as label, COUNT(`{value_based_on}`) as value
|
|
FROM `tab{source}`
|
|
GROUP BY `{based_on}`
|
|
ORDER BY value DESC
|
|
LIMIT 20
|
|
""", as_dict=True)
|
|
|
|
labels = [str(d.get("label") or "Unknown") for d in data]
|
|
values = [float(d.get("value") or 0) for d in data]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": "count", "values": values, "color": "#4F46E5"}],
|
|
"type": "Bar",
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"doctype": source},
|
|
})
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), f"Custom Chart Error: {chart.name}")
|
|
_err(f"Error processing custom chart: {str(e)}")
|
|
|
|
|
|
def _process_group_by_chart(chart):
|
|
"""
|
|
Process a Group By / Count / Sum type Dashboard Chart.
|
|
These charts aggregate data from a doctype without using a report.
|
|
"""
|
|
try:
|
|
source = chart.document_type
|
|
if not source:
|
|
_err(f"Group By chart '{chart.name}' missing document_type")
|
|
return
|
|
|
|
group_by_field = chart.group_by_based_on or chart.based_on
|
|
aggregate_function = chart.chart_type # Count, Sum, etc.
|
|
value_field = chart.aggregate_function_based_on or "name"
|
|
|
|
if not group_by_field:
|
|
_err(f"Group By chart '{chart.name}' missing group_by field")
|
|
return
|
|
|
|
# Build query based on aggregate function
|
|
if aggregate_function == "Count":
|
|
query = f"""
|
|
SELECT `{group_by_field}` as label, COUNT(*) as value
|
|
FROM `tab{source}`
|
|
WHERE `{group_by_field}` IS NOT NULL
|
|
GROUP BY `{group_by_field}`
|
|
ORDER BY value DESC
|
|
LIMIT 20
|
|
"""
|
|
elif aggregate_function == "Sum":
|
|
query = f"""
|
|
SELECT `{group_by_field}` as label, SUM(`{value_field}`) as value
|
|
FROM `tab{source}`
|
|
WHERE `{group_by_field}` IS NOT NULL
|
|
GROUP BY `{group_by_field}`
|
|
ORDER BY value DESC
|
|
LIMIT 20
|
|
"""
|
|
else:
|
|
# Default to count
|
|
query = f"""
|
|
SELECT `{group_by_field}` as label, COUNT(*) as value
|
|
FROM `tab{source}`
|
|
WHERE `{group_by_field}` IS NOT NULL
|
|
GROUP BY `{group_by_field}`
|
|
ORDER BY value DESC
|
|
LIMIT 20
|
|
"""
|
|
|
|
data = frappe.db.sql(query, as_dict=True)
|
|
|
|
labels = [str(d.get("label") or "Unknown") for d in data]
|
|
values = [float(d.get("value") or 0) for d in data]
|
|
|
|
chart_type = (chart.type or "Bar").title()
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": aggregate_function.lower(), "values": values, "color": "#6366F1"}],
|
|
"type": chart_type,
|
|
"options": _parse_custom_options(chart.custom_options),
|
|
"source": {"doctype": source},
|
|
})
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), f"Group By Chart Error: {chart.name}")
|
|
_err(f"Error processing group by chart: {str(e)}")
|
|
|
|
|
|
def _parse_custom_options(raw):
|
|
"""Parse custom_options JSON string to dict."""
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
if isinstance(raw, dict):
|
|
return raw
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_repair_cost_by_item(year=None):
|
|
"""
|
|
Example specialized endpoint for 'Repair Cost' report style chart
|
|
(X: item_code, Y: amount, Filter: Year)
|
|
"""
|
|
try:
|
|
year = int(year or frappe.utils.getdate(nowdate()).year)
|
|
rows = frappe.db.sql(
|
|
"""
|
|
SELECT item_code, SUM(amount) as amount
|
|
FROM `tabWork Order` wo
|
|
WHERE YEAR(wo.posting_date) = %(year)s
|
|
GROUP BY item_code
|
|
ORDER BY amount DESC
|
|
""",
|
|
{"year": year},
|
|
as_dict=True,
|
|
)
|
|
labels = [r.item_code or "Unknown" for r in rows]
|
|
values = [float(r.amount or 0) for r in rows]
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{"name": f"Repair Cost {year}", "values": values}],
|
|
"type": "Bar",
|
|
"options": {},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_repair_cost_by_item")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_technician_working_hours(filters=None):
|
|
"""
|
|
Dedicated endpoint for Technician Working Hours chart.
|
|
Returns hours worked by each technician.
|
|
"""
|
|
try:
|
|
if isinstance(filters, str):
|
|
filters = json.loads(filters or "{}")
|
|
filters = filters or {}
|
|
|
|
# Run the report
|
|
run = frappe.get_attr("frappe.desk.query_report.run")
|
|
report_result = run("Technicians working Hours", filters=filters)
|
|
rows = report_result.get("result", []) or []
|
|
|
|
# Filter out total rows
|
|
data_rows = [r for r in rows if isinstance(r, dict) and not r.get("is_total_row")]
|
|
|
|
labels = [str(r.get("technician_name") or "Unknown") for r in data_rows]
|
|
values = [float(r.get("total_hours") or 0) for r in data_rows]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": [{
|
|
"name": "total_hours",
|
|
"values": values,
|
|
"color": "#6366F1"
|
|
}],
|
|
"type": "Bar",
|
|
"options": {"barOptions": {"stacked": False}},
|
|
"result": data_rows,
|
|
})
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_technician_working_hours")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_technician_work_summary(filters=None):
|
|
"""
|
|
Dedicated endpoint for Technician Work Order Summary chart.
|
|
Returns work order counts (total, completed, in_progress, open) by technician.
|
|
"""
|
|
try:
|
|
if isinstance(filters, str):
|
|
filters = json.loads(filters or "{}")
|
|
filters = filters or {}
|
|
|
|
# Run the report
|
|
run = frappe.get_attr("frappe.desk.query_report.run")
|
|
report_result = run("Technician Work Order Summary", filters=filters)
|
|
rows = report_result.get("result", []) or []
|
|
|
|
# Filter out total rows
|
|
data_rows = [r for r in rows if isinstance(r, dict) and not r.get("is_total_row")]
|
|
|
|
labels = [str(r.get("assigned_technician") or "Unknown") for r in data_rows]
|
|
|
|
# Build multi-series datasets
|
|
datasets = [
|
|
{
|
|
"name": "total",
|
|
"values": [float(r.get("total") or 0) for r in data_rows],
|
|
"color": "#6366F1" # Indigo
|
|
},
|
|
{
|
|
"name": "completed",
|
|
"values": [float(r.get("completed") or 0) for r in data_rows],
|
|
"color": "#10B981" # Green
|
|
},
|
|
{
|
|
"name": "in_progress",
|
|
"values": [float(r.get("in_progress") or 0) for r in data_rows],
|
|
"color": "#3B82F6" # Blue
|
|
},
|
|
{
|
|
"name": "open",
|
|
"values": [float(r.get("open") or 0) for r in data_rows],
|
|
"color": "#F59E0B" # Amber
|
|
}
|
|
]
|
|
|
|
# Filter out datasets with all zeros
|
|
datasets = [ds for ds in datasets if any(v > 0 for v in ds["values"])]
|
|
|
|
_ok({
|
|
"labels": labels,
|
|
"datasets": datasets,
|
|
"type": "Bar",
|
|
"options": {"barOptions": {"stacked": True}},
|
|
"result": data_rows,
|
|
})
|
|
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_technician_work_summary")
|
|
_err(str(e))
|
|
|
|
|
|
def _parse_dashboard_filters(filters):
|
|
if isinstance(filters, str):
|
|
filters = json.loads(filters or "{}")
|
|
return filters or {}
|
|
|
|
|
|
def _wo_base_conditions(filters):
|
|
"""Build SQL WHERE fragments for Work_Order dashboard metrics."""
|
|
conditions = [
|
|
"(custom_delete_status IS NULL OR custom_delete_status != 'Deleted')",
|
|
"docstatus < 2",
|
|
]
|
|
values = {}
|
|
|
|
work_order_type = filters.get("work_order_type")
|
|
if work_order_type:
|
|
if isinstance(work_order_type, (list, tuple)) and len(work_order_type) == 2 and work_order_type[0] == "in":
|
|
types = work_order_type[1] or []
|
|
if types:
|
|
placeholders = ", ".join([f"%(wo_type_{i})s" for i in range(len(types))])
|
|
conditions.append(f"work_order_type IN ({placeholders})")
|
|
for i, t in enumerate(types):
|
|
values[f"wo_type_{i}"] = t
|
|
else:
|
|
conditions.append("work_order_type = %(work_order_type)s")
|
|
values["work_order_type"] = work_order_type
|
|
|
|
from_date = filters.get("from_date")
|
|
if from_date:
|
|
conditions.append("DATE(creation) >= %(from_date)s")
|
|
values["from_date"] = from_date
|
|
|
|
to_date = filters.get("to_date")
|
|
if to_date:
|
|
conditions.append("DATE(creation) <= %(to_date)s")
|
|
values["to_date"] = to_date
|
|
|
|
return " AND ".join(conditions), values
|
|
|
|
|
|
def _normalize_work_order_type_filter(value):
|
|
"""Accept Issue Type name or Frappe-style ['in', [...]] filter."""
|
|
if isinstance(value, str):
|
|
stripped = value.strip()
|
|
if stripped.startswith("["):
|
|
try:
|
|
value = json.loads(stripped)
|
|
except Exception:
|
|
return stripped
|
|
else:
|
|
return stripped
|
|
|
|
if isinstance(value, (list, tuple)) and len(value) == 2 and value[0] == "in":
|
|
return ["in", list(value[1] or [])]
|
|
|
|
return value
|
|
|
|
|
|
def _apply_department_filter(conditions, values, filters, column_name, value_prefix):
|
|
work_order_type = _normalize_work_order_type_filter(filters.get("work_order_type"))
|
|
if not work_order_type:
|
|
return
|
|
|
|
if isinstance(work_order_type, (list, tuple)) and len(work_order_type) == 2 and work_order_type[0] == "in":
|
|
types = work_order_type[1] or []
|
|
if types:
|
|
placeholders = ", ".join([f"%({value_prefix}_{i})s" for i in range(len(types))])
|
|
conditions.append(f"{column_name} IN ({placeholders})")
|
|
for i, t in enumerate(types):
|
|
values[f"{value_prefix}_{i}"] = t
|
|
else:
|
|
conditions.append(f"{column_name} = %(technical_department)s")
|
|
values["technical_department"] = work_order_type
|
|
|
|
|
|
def _asset_base_conditions(filters):
|
|
"""Build SQL WHERE fragments for Asset dashboard metrics."""
|
|
conditions = ["(custom_delete_status IS NULL OR custom_delete_status != 'Deleted')"]
|
|
values = {}
|
|
|
|
_apply_department_filter(
|
|
conditions,
|
|
values,
|
|
filters,
|
|
"custom_technical_department",
|
|
"asset_dept",
|
|
)
|
|
|
|
from_date = filters.get("from_date")
|
|
if from_date:
|
|
conditions.append("DATE(creation) >= %(from_date)s")
|
|
values["from_date"] = from_date
|
|
|
|
to_date = filters.get("to_date")
|
|
if to_date:
|
|
conditions.append("DATE(creation) <= %(to_date)s")
|
|
values["to_date"] = to_date
|
|
|
|
return " AND ".join(conditions), values
|
|
|
|
|
|
def _normalize_repair_status(status):
|
|
if not status:
|
|
return "Unknown"
|
|
s = str(status).lower().strip()
|
|
if "open" in s:
|
|
return "Open"
|
|
if "work in progress" in s or "in progress" in s or "wip" in s:
|
|
return "Work In Progress"
|
|
if "pending review" in s or "pending" in s:
|
|
return "Pending Review"
|
|
if "completed" in s or "complete" in s:
|
|
return "Completed"
|
|
if "closed" in s:
|
|
return "Closed"
|
|
if "rejected" in s:
|
|
return "Rejected"
|
|
return str(status)
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_dashboard_work_order_metrics(filters=None):
|
|
"""
|
|
Aggregated Work Order metrics for ModernDashboard (no full-table fetch).
|
|
"""
|
|
try:
|
|
filters = _parse_dashboard_filters(filters)
|
|
where_sql, values = _wo_base_conditions(filters)
|
|
|
|
rows = frappe.db.sql(
|
|
f"""
|
|
SELECT
|
|
work_order_type,
|
|
repair_status,
|
|
workflow_state,
|
|
COUNT(name) AS cnt
|
|
FROM `tabWork_Order`
|
|
WHERE {where_sql}
|
|
GROUP BY work_order_type, repair_status, workflow_state
|
|
""",
|
|
values,
|
|
as_dict=True,
|
|
)
|
|
|
|
status_counts = {}
|
|
open_count = in_progress_count = completed_count = 0
|
|
pending_review_count = rejected_count = 0
|
|
wf_completed_count = wf_in_progress_count = wf_rejected_count = 0
|
|
grouped_data = {}
|
|
all_statuses = set()
|
|
type_stat_map = {}
|
|
|
|
for row in rows:
|
|
cnt = int(row.cnt or 0)
|
|
repair_status = row.repair_status or "Unknown"
|
|
status_counts[repair_status] = status_counts.get(repair_status, 0) + cnt
|
|
|
|
sl = str(repair_status).lower()
|
|
if sl == "open":
|
|
open_count += cnt
|
|
if sl == "work in progress":
|
|
in_progress_count += cnt
|
|
if sl == "closed":
|
|
completed_count += cnt
|
|
if sl == "pending review":
|
|
pending_review_count += cnt
|
|
if sl == "rejected":
|
|
rejected_count += cnt
|
|
|
|
wf_state = (row.workflow_state or "").lower()
|
|
if wf_state in ("closed", "sent to supervisor"):
|
|
wf_completed_count += cnt
|
|
elif wf_state == "rejected":
|
|
wf_rejected_count += cnt
|
|
else:
|
|
wf_in_progress_count += cnt
|
|
|
|
raw_type = row.work_order_type
|
|
if not raw_type or str(raw_type).strip().lower() in ("", "null", "unknown", "undefined", "none"):
|
|
continue
|
|
|
|
work_order_type = str(raw_type)
|
|
normalized = _normalize_repair_status(repair_status)
|
|
all_statuses.add(normalized)
|
|
if work_order_type not in grouped_data:
|
|
grouped_data[work_order_type] = {}
|
|
grouped_data[work_order_type][normalized] = grouped_data[work_order_type].get(normalized, 0) + cnt
|
|
|
|
if work_order_type not in type_stat_map:
|
|
type_stat_map[work_order_type] = {
|
|
"total": 0,
|
|
"wfCompleted": 0,
|
|
"wfInProgress": 0,
|
|
"wfRejected": 0,
|
|
"closed": 0,
|
|
"pendingReview": 0,
|
|
"open": 0,
|
|
"inProgress": 0,
|
|
"rejected": 0,
|
|
}
|
|
s = type_stat_map[work_order_type]
|
|
s["total"] += cnt
|
|
if wf_state in ("closed", "sent to supervisor"):
|
|
s["wfCompleted"] += cnt
|
|
elif wf_state == "rejected":
|
|
s["wfRejected"] += cnt
|
|
else:
|
|
s["wfInProgress"] += cnt
|
|
if sl == "closed":
|
|
s["closed"] += cnt
|
|
if sl == "pending review":
|
|
s["pendingReview"] += cnt
|
|
if sl == "open":
|
|
s["open"] += cnt
|
|
if sl == "work in progress":
|
|
s["inProgress"] += cnt
|
|
if sl == "rejected":
|
|
s["rejected"] += cnt
|
|
|
|
total = sum(status_counts.values())
|
|
status_display_order = [
|
|
"Open", "Work In Progress", "Pending Review", "Closed", "Rejected", "Cancelled", "Unknown",
|
|
]
|
|
sorted_entries = sorted(
|
|
status_counts.items(),
|
|
key=lambda item: (
|
|
status_display_order.index(item[0])
|
|
if item[0] in status_display_order
|
|
else len(status_display_order)
|
|
),
|
|
)
|
|
labels = [label for label, _ in sorted_entries]
|
|
pie_values = [count for _, count in sorted_entries]
|
|
color_map = {
|
|
"Open": "#F59E0B",
|
|
"Work In Progress": "#3B82F6",
|
|
"Pending Review": "#E11D48",
|
|
"Pending": "#8B5CF6",
|
|
"Completed": "#10B981",
|
|
"Closed": "#10B981",
|
|
"Rejected": "#EF4444",
|
|
"Cancelled": "#EC4899",
|
|
"Unknown": "#6B7280",
|
|
}
|
|
pie_colors = [color_map.get(label, "#6366F1") for label in labels]
|
|
|
|
work_order_type_labels = list(grouped_data.keys())
|
|
status_order = ["Open", "Work In Progress", "Pending Review", "Completed", "Closed", "Rejected"]
|
|
status_list = sorted(
|
|
all_statuses,
|
|
key=lambda a: (
|
|
(status_order.index(a), a)
|
|
if a in status_order
|
|
else (len(status_order), a)
|
|
),
|
|
)
|
|
grouped_datasets = [
|
|
{
|
|
"name": status,
|
|
"values": [grouped_data[t].get(status, 0) for t in work_order_type_labels],
|
|
}
|
|
for status in status_list
|
|
]
|
|
|
|
completion_type_labels = list(type_stat_map.keys())
|
|
completion_rates = [
|
|
{
|
|
"type": t,
|
|
**type_stat_map[t],
|
|
"rate": (type_stat_map[t]["wfCompleted"] / type_stat_map[t]["total"] * 100)
|
|
if type_stat_map[t]["total"] > 0
|
|
else 0,
|
|
}
|
|
for t in completion_type_labels
|
|
]
|
|
|
|
avg_row = frappe.db.sql(
|
|
f"""
|
|
SELECT AVG(TIMESTAMPDIFF(HOUR, creation, first_responded_on)) AS avg_hours
|
|
FROM `tabWork_Order`
|
|
WHERE {where_sql}
|
|
AND first_responded_on IS NOT NULL
|
|
AND first_responded_on >= creation
|
|
""",
|
|
values,
|
|
as_dict=True,
|
|
)
|
|
avg_response_time = float(avg_row[0].avg_hours or 0) if avg_row else 0
|
|
|
|
_ok({
|
|
"counts": {
|
|
"open": open_count,
|
|
"inProgress": in_progress_count,
|
|
"completed": completed_count,
|
|
"pendingReview": pending_review_count,
|
|
"rejected": rejected_count,
|
|
"total": total,
|
|
"wfCompleted": wf_completed_count,
|
|
"wfInProgress": wf_in_progress_count,
|
|
"wfRejected": wf_rejected_count,
|
|
},
|
|
"work_order_chart": {
|
|
"labels": labels,
|
|
"datasets": [{"name": "Work Orders", "values": pie_values, "colors": pie_colors}],
|
|
"type": "bar",
|
|
},
|
|
"work_order_grouped_chart": {
|
|
"labels": work_order_type_labels,
|
|
"datasets": grouped_datasets,
|
|
"type": "Bar",
|
|
},
|
|
"completion_by_type": {
|
|
"labels": completion_type_labels,
|
|
"rates": completion_rates,
|
|
},
|
|
"avg_response_time": avg_response_time,
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_dashboard_work_order_metrics")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_asset_device_status_summary(filters=None):
|
|
"""Aggregated asset device status counts for dashboard pie chart."""
|
|
try:
|
|
filters = _parse_dashboard_filters(filters)
|
|
where_sql, values = _asset_base_conditions(filters)
|
|
|
|
rows = frappe.db.sql(
|
|
f"""
|
|
SELECT COALESCE(NULLIF(TRIM(custom_device_status), ''), 'Unknown') AS status_label,
|
|
COUNT(name) AS cnt
|
|
FROM `tabAsset`
|
|
WHERE {where_sql}
|
|
GROUP BY status_label
|
|
ORDER BY cnt DESC
|
|
""",
|
|
values,
|
|
as_dict=True,
|
|
)
|
|
|
|
labels = [r.status_label for r in rows]
|
|
values_list = [int(r.cnt or 0) for r in rows]
|
|
total = sum(values_list)
|
|
color_palette = {
|
|
"Up": "#10B981",
|
|
"up": "#10B981",
|
|
"Down": "#EF4444",
|
|
"down": "#EF4444",
|
|
"Under Maintenance": "#F59E0B",
|
|
"under maintenance": "#F59E0B",
|
|
"Decommissioned": "#6B7280",
|
|
}
|
|
default_colors = ["#6366F1", "#8B5CF6", "#06B6D4", "#EC4899", "#F59E0B", "#10B981"]
|
|
colors = [
|
|
color_palette.get(lbl) or color_palette.get(lbl.lower()) or default_colors[i % len(default_colors)]
|
|
for i, lbl in enumerate(labels)
|
|
]
|
|
assets_down = sum(
|
|
int(r.cnt or 0)
|
|
for r in rows
|
|
if str(r.status_label).strip().lower() == "down"
|
|
)
|
|
|
|
_ok({
|
|
"total_assets": total,
|
|
"assets_down": assets_down,
|
|
"chart": {
|
|
"labels": labels,
|
|
"datasets": [{"name": "Assets", "values": values_list, "colors": colors}],
|
|
"type": "Pie",
|
|
"total": total,
|
|
},
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_asset_device_status_summary")
|
|
_err(str(e))
|
|
|
|
|
|
def _maintenance_log_base_conditions(filters):
|
|
"""Build SQL WHERE fragments for Asset Maintenance Log dashboard metrics."""
|
|
conditions = ["docstatus < 2"]
|
|
values = {}
|
|
|
|
_apply_department_filter(
|
|
conditions,
|
|
values,
|
|
filters,
|
|
"custom_technical_department",
|
|
"aml_dept",
|
|
)
|
|
|
|
from_date = filters.get("from_date")
|
|
if from_date:
|
|
conditions.append("DATE(due_date) >= %(from_date)s")
|
|
values["from_date"] = from_date
|
|
|
|
to_date = filters.get("to_date")
|
|
if to_date:
|
|
conditions.append("DATE(due_date) <= %(to_date)s")
|
|
values["to_date"] = to_date
|
|
|
|
return " AND ".join(conditions), values
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def get_maintenance_insights(filters=None):
|
|
"""Aggregated maintenance log metrics without loading all logs."""
|
|
try:
|
|
filters = _parse_dashboard_filters(filters)
|
|
where_sql, values = _maintenance_log_base_conditions(filters)
|
|
|
|
row = frappe.db.sql(
|
|
f"""
|
|
SELECT
|
|
COUNT(name) AS total,
|
|
SUM(CASE WHEN LOWER(maintenance_status) = 'completed' THEN 1 ELSE 0 END) AS completed,
|
|
SUM(CASE WHEN LOWER(maintenance_status) = 'planned' THEN 1 ELSE 0 END) AS planned,
|
|
SUM(
|
|
CASE
|
|
WHEN due_date IS NOT NULL
|
|
AND due_date < CURDATE()
|
|
AND LOWER(COALESCE(maintenance_status, '')) != 'completed'
|
|
AND LOWER(COALESCE(maintenance_status, '')) != 'cancelled'
|
|
THEN 1 ELSE 0
|
|
END
|
|
) AS overdue
|
|
FROM `tabAsset Maintenance Log`
|
|
WHERE {where_sql}
|
|
""",
|
|
values,
|
|
as_dict=True,
|
|
)
|
|
data = row[0] if row else {}
|
|
total = int(data.total or 0)
|
|
completed = int(data.completed or 0)
|
|
planned = int(data.planned or 0)
|
|
overdue = int(data.overdue or 0)
|
|
efficiency = (completed / total * 100) if total > 0 else 0
|
|
|
|
_ok({
|
|
"total": total,
|
|
"completed": completed,
|
|
"planned": planned,
|
|
"overdue": overdue,
|
|
"maintenance_efficiency": efficiency,
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "get_maintenance_insights")
|
|
_err(str(e))
|
|
|
|
|
|
@frappe.whitelist(allow_guest=True)
|
|
def debug_list_charts():
|
|
"""
|
|
Debug endpoint to list all Dashboard Chart names.
|
|
Use this to verify exact chart names in the database.
|
|
"""
|
|
try:
|
|
charts = frappe.db.sql("""
|
|
SELECT name, chart_name, chart_type, report_name, is_public, type
|
|
FROM `tabDashboard Chart`
|
|
ORDER BY modified DESC
|
|
""", as_dict=True)
|
|
|
|
_ok({
|
|
"total": len(charts),
|
|
"charts": charts
|
|
})
|
|
except Exception as e:
|
|
frappe.log_error(frappe.get_traceback(), "debug_list_charts")
|
|
_err(str(e)) |