Untitled
unknown
plain_text
6 days ago
10 kB
2
Indexable
# Copyright (c) 2025, haron computer and contributors # For license information, please see license.txt import frappe from frappe import _ from datetime import datetime, timedelta from frappe.query_builder import DocType from frappe.query_builder.functions import IfNull, Length, Date import frappe.utils invalid_reasons = { "unsubmitted_appointment": "Unsubmitted Machine Appointment date has passed!", "fm_spare_not_in_problemlist": "Requested FM Spare Part is not in Problem List!", "fm_reason_missing": "FM problem exists but reason is missing!", "spare_request_empty": "Spare part request should not be empty!", "invalid_mobile": "Invalid E mobile number!", "invalid_encode_date": "Encode date should be same date or later than Maintenance finished date!", "mobile_iccid_mismatch": "E mobile no & ICCID no should be filled or not filled not filled together!", "annual_service_date_invalid": "Next annual service date should be greater than or Equal to January 2025!", "tax_officer_required": "Tax officer name is required unless is mini is true!", } def execute(filters: dict | None = {}): data = get_data(filters=filters) columns = get_columns() return columns, data def get_data(filters: dict | None = {}): invalid_maintenances = [] def add_to_invalid_maintenances(row, reason): invalid_maintenances.append( { "machine_reception_name": row.machine_reception, "mrc": row.machine_number, "responsible_person": f"{row.employee_name}", "date": row.received_at, "reason": reason, } ) Maintenance = DocType("Machine Maintenance") Reception = DocType("Machine Reception") Technician = DocType("Technician") Employee = DocType("Employee") query = ( frappe.qb.from_(Maintenance) .join(Reception) .on(Reception.name == Maintenance.machine_reception) .join(Technician) .on(Technician.name == Maintenance.technician) .join(Employee) .on(Technician.employee == Employee.name) .select( Maintenance.name, Maintenance.machine_reception, Maintenance.machine_number, Reception.received_at, Employee.name, Employee.employee_name, ) ) from_date_filter = filters.get("from_date") to_date_filter = filters.get("to_date") format_date = lambda date: datetime.strptime(date, "%Y-%m-%d").date() if date else None employee_filter = filters.get("employee") if employee_filter: query = query.where(Employee.name == employee_filter) if from_date_filter and to_date_filter: from_date = format_date(from_date_filter) to_date = format_date(to_date_filter) query = query.where(Reception.received_at[from_date:to_date]) elif from_date_filter: from_date = format_date(from_date_filter) query = query.where(Reception.received_at >= from_date) elif to_date_filter: to_date = format_date(to_date_filter) query = query.where(Reception.received_at <= to_date) def filter_reason(reason: str): return not filters.get("reason") or filters.get("reason") == invalid_reasons.get(reason) def sql_req(sql): format_date = lambda date: ( datetime.strptime(date, "%Y-%m-%d") if len(date) == 10 else datetime.strptime(date, "%Y-%m-%d %H:%M:%S") ) params = {"today": today, "next_annual_service_valid_date": datetime(year=2025, month=1, day=31)} if from_date_filter and to_date_filter: params["from_date"] = format_date(from_date_filter) params["to_date"] = format_date(to_date_filter) elif from_date_filter: params["from_date"] = format_date(from_date_filter) elif to_date_filter: params["to_date"] = format_date(to_date_filter) return frappe.db.sql(sql, params, as_dict=True) # Unsubmitted maintenances past appointment date if filter_reason("unsubmitted_appointment"): today = datetime.now().date() + timedelta(days=10) query = query.where(Reception.appointment_date < today).where(Maintenance.workflow_state != "Submitted") for row in query.run(as_dict=True): reason = invalid_reasons.get("unsubmitted_appointment", "") add_to_invalid_maintenances(row, reason) # FM Spare requested but not in problem list if filter_reason("fm_spare_not_in_problemlist"): RequestedItemList = DocType("Requested Item List") FaultLink = DocType("Fault Link") subquery = ( frappe.qb.from_(RequestedItemList) .join(FaultLink) .on(FaultLink.parent == RequestedItemList.parent) .where(FaultLink.fault != "FM CHANGE") .select(RequestedItemList.parent) ) query = query.where(Maintenance.name.notin(subquery)) for row in query.run(as_dict=True): reason = invalid_reasons.get("fm_spare_not_in_problemlist", "") add_to_invalid_maintenances(row, reason) # FM Change in problem list but missing reason if filter_reason("fm_reason_missing"): query = query.where(IfNull(Maintenance.fm_change_reason, "") == "") for row in query.run(as_dict=True): reason = invalid_reasons.get("fm_reason_missing", "") add_to_invalid_maintenances(row, reason) # Missing spare part requests if filter_reason("spare_request_empty"): RequestedItemList = DocType("Requested Item List") subquery = ( frappe.qb.from_(RequestedItemList) .select(RequestedItemList.parent) .distinct() ) query = query.where(Maintenance.name.notin(subquery)) for row in query.run(as_dict=True): reason = invalid_reasons.get("spare_request_empty", "") add_to_invalid_maintenances(row, reason) # Invalid E mobile number if filter_reason("invalid_mobile"): query = query.where( (Maintenance.e_mobile_no.not_like("09%") | (Length(Maintenance.e_mobile_no) != 10)) & (Maintenance.e_mobile_no.not_like("+251%") | (Length(Maintenance.e_mobile_no) != 13)) ) for row in query.run(as_dict=True): reason = invalid_reasons.get("invalid_mobile", "") add_to_invalid_maintenances(row, reason) # Maintenance finished date >= encode date if filter_reason("invalid_encode_date"): query = query.where(Maintenance.maintenance_finished_date >= Maintenance.encoded_at) for row in query.run(as_dict=True): reason = invalid_reasons.get("invalid_encode_date", "") add_to_invalid_maintenances(row, reason) # E mobile no & iccid no should be filled or not filled together if filter_reason("mobile_iccid_mismatch"): query = query.where( ~( (Maintenance.e_mobile_no.isnull() & Maintenance.iccid_no.isnull()) | (Maintenance.e_mobile_no.isnotnull() & Maintenance.iccid_no.isnotnull()) ) ) for row in query.run(as_dict=True): reason = invalid_reasons.get("mobile_iccid_mismatch", "") add_to_invalid_maintenances(row, reason) # next annual service date < 2025, first month if filter_reason("annual_service_date_invalid"): query = query.where(Maintenance.next_annual_service_date > datetime(year=2025, month=1, day=31).date()) for row in query.run(as_dict=True): reason = invalid_reasons.get("annual_service_date_invalid") add_to_invalid_maintenances(row, reason) # Tax officer name is required except for is mini service if filter_reason("tax_officer_required"): query = query.where((Maintenance.is_mini == 0) & Maintenance.tax_officer_name.isnotnull()) for row in query.run(as_dict=True): reason = invalid_reasons.get("tax_officer_required") add_to_invalid_maintenances(row, reason) data = [] for maintenance in invalid_maintenances: data.append( { "machine_reception_name": maintenance.get("machine_reception_name", ""), "mrc": maintenance.get("mrc", ""), "responsible_person": maintenance.get("responsible_person", ""), "date": maintenance.get("date", ""), "reason": maintenance.get("reason", ""), } ) return data def get_filters(filters: dict | None = {}): filter_opts = {} if filters.get("reason"): filter_opts["reason"] = filters["reason"] if filters.get("from_date") and filters.get("to_date"): filter_opts["date_and_time"] = ["between", [filters["from_date"], filters["to_date"]]] elif filters.get("from_date"): filter_opts["date_and_time"] = [">=", filters["from_date"]] elif filters.get("to_date"): filter_opts["date_and_time"] = ["<=", filters["to_date"]] return filter_opts or filters def get_columns(): return [ { "fieldname": "machine_reception_name", "label": _("Machine Reception Name"), "fieldtype": "Data", }, { "fieldname": "mrc", "label": _("MRC"), "fieldtype": "Data", }, { "fieldname": "responsible_person", "label": _("Responsible Person"), "fieldtype": "Data", }, { "fieldname": "date", "label": _("Date (Received at)"), "fieldtype": "Date", }, { "fieldname": "reason", "label": _("Reason"), "fieldtype": "Data", "width": 500, }, ] # @frappe.whitelist() # def get_invalid_reasons(): # return invalid_reasons
Editor is loading...
Leave a Comment