Iman_Asset_Lite/asset_lite/api/wo_feedback_api.py

143 lines
3.8 KiB
Python

import frappe
from frappe import _
RATING_ORDER = [
"Excellent / ممتاز",
"Good / جيد",
"Average / متوسط",
"Poor / ضعيف",
]
REPORT_COLUMNS = [
{"label": _("Work Order"), "fieldname": "work_order", "fieldtype": "Link", "options": "Work_Order", "width": 150},
{"label": _("Requester Name"), "fieldname": "requester_name", "fieldtype": "Data", "width": 180},
{"label": _("Rating"), "fieldname": "rating", "fieldtype": "Data", "width": 160},
{"label": _("Comments"), "fieldname": "comments", "fieldtype": "Small Text", "width": 220},
{"label": _("Created On"), "fieldname": "creation", "fieldtype": "Datetime", "width": 150},
{"label": _("Last Modified"), "fieldname": "modified", "fieldtype": "Datetime", "width": 150},
]
def _parse_filters(filters=None):
if isinstance(filters, str):
filters = frappe.parse_json(filters or "{}")
return filters or {}
def _build_wo_feedback_conditions(filters):
conditions = ["IFNULL(wf.rating, '') != ''"]
params = {}
if filters.get("work_order"):
conditions.append("wf.work_order = %(work_order)s")
params["work_order"] = filters.get("work_order")
if filters.get("requester_name"):
conditions.append("wf.requester_name LIKE %(requester_name)s")
params["requester_name"] = f"%{filters.get('requester_name')}%"
if filters.get("rating"):
conditions.append("wf.rating = %(rating)s")
params["rating"] = filters.get("rating")
if filters.get("company"):
conditions.append("wo.company = %(company)s")
params["company"] = filters.get("company")
if filters.get("site_name"):
conditions.append("wo.site_name = %(site_name)s")
params["site_name"] = filters.get("site_name")
return conditions, params
def _rating_sort_key(rating):
try:
return RATING_ORDER.index(rating)
except ValueError:
return len(RATING_ORDER)
@frappe.whitelist(allow_guest=True)
def get_wo_feedback_for_work_order(work_order: str | None = None):
"""Guest-safe read for WO Feedback (autoname = work_order field value)."""
if not work_order or not str(work_order).strip():
frappe.throw(_("Work Order is required"))
work_order = str(work_order).strip()
feedback_name = work_order
if not frappe.db.exists("WO Feedback", work_order):
feedback_name = frappe.db.get_value("WO Feedback", {"work_order": work_order}, "name")
if not feedback_name:
return None
return frappe.db.get_value(
"WO Feedback",
feedback_name,
["name", "work_order", "requester_name", "rating", "comments"],
as_dict=True,
)
@frappe.whitelist()
def run_wo_feedback_report(filters=None):
"""Return WO feedback detail rows for dashboard popup (supports hospital/site filters)."""
filters = _parse_filters(filters)
conditions, params = _build_wo_feedback_conditions(filters)
where = " AND ".join(conditions)
rows = frappe.db.sql(
f"""
SELECT
wf.work_order,
wf.requester_name,
wf.rating,
wf.comments,
wf.creation,
wf.modified
FROM `tabWO Feedback` wf
INNER JOIN `tabWork_Order` wo ON wo.name = wf.work_order
WHERE {where}
ORDER BY wf.modified DESC
""",
params,
as_dict=True,
)
return {
"columns": REPORT_COLUMNS,
"result": rows,
}
@frappe.whitelist()
def get_wo_feedback_summary(filters=None):
"""Return chart-ready rating counts for dashboard card."""
filters = _parse_filters(filters)
conditions, params = _build_wo_feedback_conditions(filters)
where = " AND ".join(conditions)
rows = frappe.db.sql(
f"""
SELECT wf.rating, COUNT(*) AS count
FROM `tabWO Feedback` wf
INNER JOIN `tabWork_Order` wo ON wo.name = wf.work_order
WHERE {where}
GROUP BY wf.rating
""",
params,
as_dict=True,
)
count_by_rating = {row.rating: int(row.count or 0) for row in rows if row.rating}
labels = RATING_ORDER[:]
values = [count_by_rating.get(label, 0) for label in labels]
return {
"labels": labels,
"datasets": [{"name": "Responses", "values": values}],
"type": "Bar",
}