import frappe from frappe import _ # Default fields for PM Schedule Generator doctype PM_SCHEDULE_GENERATOR_FIELDS = [ 'name', 'owner', 'creation', 'modified', 'modified_by', 'docstatus', 'idx', 'hospital', 'modality', 'device_status', 'start_date', 'maintenance_team', 'maintenance_manager', 'end_date', 'periodicity', 'assign_to', 'due_date' ] # Child table: PM Entry Line (maintenance_entries) PM_ENTRY_LINE_FIELDS = [ 'name', 'owner', 'creation', 'modified', 'modified_by', 'docstatus', 'idx', 'parent', 'parentfield', 'parenttype', 'asset', 'asset_name', 'start_date', 'end_date', 'manufacturer', 'model' ] @frappe.whitelist() def create_bulk_schedules(asset_names, start_date, end_date, maintenance_team=None, periodicity='Monthly', maintenance_type='Preventive'): """Create maintenance schedules for multiple assets""" created_schedules = [] for asset_name in asset_names: # Create Asset Maintenance record maintenance = frappe.get_doc({ 'doctype': 'Asset Maintenance', 'asset_name': asset_name, 'maintenance_type': maintenance_type, 'periodicity': periodicity, 'maintenance_team': maintenance_team }) maintenance.insert() created_schedules.append(maintenance.name) return { 'success': True, 'created': len(created_schedules), 'schedules': created_schedules } def get_child_table_data(parent_name, parentfield, child_doctype, fields=None): """ Get child table data for a PM Schedule Generator Args: parent_name: Name of the parent PM Schedule Generator parentfield: Field name of the child table in parent child_doctype: Doctype of the child table fields: List of fields to return Returns: List of child table records """ try: return frappe.get_all( child_doctype, filters={ 'parent': parent_name, 'parentfield': parentfield, 'parenttype': 'PM Schedule Generator' }, fields=fields or ['*'], order_by='idx asc' ) except Exception: return [] @frappe.whitelist(allow_guest=True) def get_pm_schedules(filters=None, fields=None, limit=20, offset=0, order_by=None, include_child_tables=False): """ Get list of PM Schedule Generators with filters and pagination Args: filters: JSON string of filters (e.g., '{"hospital": "Domat Al Jandal Hospital"}') fields: JSON string of fields to return (e.g., '["hospital", "modality"]') limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) order_by: Sort order (e.g., "creation desc") include_child_tables: Whether to include child table data (default: False) Returns: { "pm_schedules": [...], "total_count": int, "limit": int, "offset": int, "has_more": bool } """ try: import json # Parse filters if provided if filters and isinstance(filters, str): filters = json.loads(filters) # Parse fields if provided if fields and isinstance(fields, str): fields = json.loads(fields) else: fields = PM_SCHEDULE_GENERATOR_FIELDS.copy() # Parse include_child_tables if isinstance(include_child_tables, str): include_child_tables = include_child_tables.lower() in ('true', '1', 'yes') # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters or {}) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters or {}, fields=fields, limit_page_length=int(limit), limit_start=int(offset), order_by=order_by or 'creation desc' ) # Include child tables if requested if include_child_tables: for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_pm_schedule_details(pm_schedule_name): """ Get detailed information about a specific PM Schedule Generator Args: pm_schedule_name: Name/ID of the PM Schedule Generator Returns: PM Schedule Generator document with all fields including child tables """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Check if user has permission to read this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'read', pm_schedule_name): frappe.throw(_('Not permitted to access this PM Schedule Generator')) # Get PM Schedule details pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) # Convert to dict and include child tables pm_schedule_dict = pm_schedule.as_dict() # Ensure child tables are included with all fields pm_schedule_dict['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] frappe.response['message'] = pm_schedule_dict except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedule Details API Error') frappe.response['message'] = { 'error': str(e) } @frappe.whitelist(allow_guest=True) def create_pm_schedule(pm_schedule_data): """ Create a new PM Schedule Generator Args: pm_schedule_data: JSON string containing PM Schedule fields including child tables Example: { "hospital": "Domat Al Jandal Hospital", "modality": "X-Ray", "start_date": "2025-12-23", "end_date": "2026-12-23", "periodicity": "Monthly", "maintenance_team": "DAJH Maintenance Team", "maintenance_manager": "manager@example.com", "assign_to": "technician@example.com", "due_date": "2026-01-23", "maintenance_entries": [ { "asset": "ACC-ASS-2025-00100", "asset_name": "Test Asset 1", "start_date": "2025-12-23", "end_date": "2026-12-23", "manufacturer": "ABV", "model": "" } ] } Returns: Created PM Schedule Generator document """ try: import json # Parse PM schedule data if isinstance(pm_schedule_data, str): pm_schedule_data = json.loads(pm_schedule_data) # Check if user has permission to create PM Schedule if not frappe.has_permission('PM Schedule Generator', 'create'): frappe.throw(_('Not permitted to create PM Schedule Generator')) # Create new PM Schedule pm_schedule = frappe.get_doc({ 'doctype': 'PM Schedule Generator', **pm_schedule_data }) pm_schedule.insert() frappe.db.commit() # Return created PM Schedule with child tables pm_schedule_dict = pm_schedule.as_dict() pm_schedule_dict['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] frappe.response['message'] = { 'success': True, 'pm_schedule': pm_schedule_dict, 'message': _('PM Schedule Generator created successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Create PM Schedule API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def update_pm_schedule(pm_schedule_name, pm_schedule_data): """ Update an existing PM Schedule Generator including child tables Args: pm_schedule_name: Name/ID of the PM Schedule Generator pm_schedule_data: JSON string containing fields to update Example: { "periodicity": "Quarterly", "maintenance_entries": [ { "asset": "ACC-ASS-2025-00100", "asset_name": "Test Asset 1", "start_date": "2025-12-23", "end_date": "2026-12-23" } ] } Returns: Updated PM Schedule Generator document """ try: import json if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Parse PM schedule data if isinstance(pm_schedule_data, str): pm_schedule_data = json.loads(pm_schedule_data) # Check if user has permission to update this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'write', pm_schedule_name): frappe.throw(_('Not permitted to update this PM Schedule Generator')) # Get PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) # Handle child tables separately child_tables = ['maintenance_entries'] for key, value in pm_schedule_data.items(): if key in child_tables: # Clear existing child table entries and add new ones pm_schedule.set(key, []) for item in value: pm_schedule.append(key, item) elif hasattr(pm_schedule, key): setattr(pm_schedule, key, value) pm_schedule.save() frappe.db.commit() # Return updated PM Schedule with child tables pm_schedule_dict = pm_schedule.as_dict() pm_schedule_dict['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] frappe.response['message'] = { 'success': True, 'pm_schedule': pm_schedule_dict, 'message': _('PM Schedule Generator updated successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Update PM Schedule API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def delete_pm_schedule(pm_schedule_name): """ Delete a PM Schedule Generator Args: pm_schedule_name: Name/ID of the PM Schedule Generator Returns: Success message """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Check if user has permission to delete this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'delete', pm_schedule_name): frappe.throw(_('Not permitted to delete this PM Schedule Generator')) # Delete PM Schedule (child tables will be deleted automatically) frappe.delete_doc('PM Schedule Generator', pm_schedule_name) frappe.db.commit() frappe.response['message'] = { 'success': True, 'message': _('PM Schedule Generator deleted successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Delete PM Schedule API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def submit_pm_schedule(pm_schedule_name): """ Submit a PM Schedule Generator (change docstatus to 1) Args: pm_schedule_name: Name/ID of the PM Schedule Generator Returns: Submitted PM Schedule Generator document """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Check if user has permission to submit this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'submit', pm_schedule_name): frappe.throw(_('Not permitted to submit this PM Schedule Generator')) # Get and submit PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) pm_schedule.submit() frappe.db.commit() # Return submitted PM Schedule with child tables pm_schedule_dict = pm_schedule.as_dict() pm_schedule_dict['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] frappe.response['message'] = { 'success': True, 'pm_schedule': pm_schedule_dict, 'message': _('PM Schedule Generator submitted successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Submit PM Schedule API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def cancel_pm_schedule(pm_schedule_name): """ Cancel a PM Schedule Generator (change docstatus to 2) Args: pm_schedule_name: Name/ID of the PM Schedule Generator Returns: Cancelled PM Schedule Generator document """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Check if user has permission to cancel this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'cancel', pm_schedule_name): frappe.throw(_('Not permitted to cancel this PM Schedule Generator')) # Get and cancel PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) pm_schedule.cancel() frappe.db.commit() # Return cancelled PM Schedule with child tables pm_schedule_dict = pm_schedule.as_dict() pm_schedule_dict['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] frappe.response['message'] = { 'success': True, 'pm_schedule': pm_schedule_dict, 'message': _('PM Schedule Generator cancelled successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Cancel PM Schedule API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def add_maintenance_entry(pm_schedule_name, entry_data): """ Add a maintenance entry to PM Schedule's maintenance_entries child table Args: pm_schedule_name: Name/ID of the PM Schedule Generator entry_data: JSON string containing maintenance entry fields Example: { "asset": "ACC-ASS-2025-00100", "asset_name": "Test Asset 1", "start_date": "2025-12-23", "end_date": "2026-12-23", "manufacturer": "ABV", "model": "Model X" } Returns: Updated PM Schedule with maintenance entries """ try: import json if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Parse entry data if isinstance(entry_data, str): entry_data = json.loads(entry_data) # Check if user has permission to update this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'write', pm_schedule_name): frappe.throw(_('Not permitted to update this PM Schedule Generator')) # Get PM Schedule and add maintenance entry pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) pm_schedule.append('maintenance_entries', entry_data) pm_schedule.save() frappe.db.commit() frappe.response['message'] = { 'success': True, 'maintenance_entries': [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])], 'message': _('Maintenance entry added successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Add Maintenance Entry API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def remove_maintenance_entry(pm_schedule_name, entry_name): """ Remove a maintenance entry from PM Schedule's maintenance_entries child table Args: pm_schedule_name: Name/ID of the PM Schedule Generator entry_name: Name/ID of the maintenance entry to remove Returns: Updated PM Schedule with remaining maintenance entries """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) if not entry_name: frappe.throw(_('Maintenance entry name is required')) # Check if user has permission to update this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'write', pm_schedule_name): frappe.throw(_('Not permitted to update this PM Schedule Generator')) # Get PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) # Find and remove the maintenance entry entry_to_remove = None for entry in pm_schedule.maintenance_entries: if entry.name == entry_name: entry_to_remove = entry break if entry_to_remove: pm_schedule.remove(entry_to_remove) pm_schedule.save() frappe.db.commit() frappe.response['message'] = { 'success': True, 'maintenance_entries': [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])], 'message': _('Maintenance entry removed successfully') } else: frappe.response['message'] = { 'success': False, 'error': _('Maintenance entry not found') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Remove Maintenance Entry API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def update_maintenance_entry(pm_schedule_name, entry_name, entry_data): """ Update a specific maintenance entry in the PM Schedule Args: pm_schedule_name: Name/ID of the PM Schedule Generator entry_name: Name/ID of the maintenance entry to update entry_data: JSON string containing fields to update Returns: Updated PM Schedule with maintenance entries """ try: import json if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) if not entry_name: frappe.throw(_('Maintenance entry name is required')) # Parse entry data if isinstance(entry_data, str): entry_data = json.loads(entry_data) # Check if user has permission to update this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'write', pm_schedule_name): frappe.throw(_('Not permitted to update this PM Schedule Generator')) # Get PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) # Find and update the maintenance entry entry_found = False for entry in pm_schedule.maintenance_entries: if entry.name == entry_name: for key, value in entry_data.items(): if hasattr(entry, key): setattr(entry, key, value) entry_found = True break if entry_found: pm_schedule.save() frappe.db.commit() frappe.response['message'] = { 'success': True, 'maintenance_entries': [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])], 'message': _('Maintenance entry updated successfully') } else: frappe.response['message'] = { 'success': False, 'error': _('Maintenance entry not found') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Update Maintenance Entry API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def get_pm_schedule_child_tables(pm_schedule_name, child_table=None): """ Get child table data for a PM Schedule Generator Args: pm_schedule_name: Name/ID of the PM Schedule Generator child_table: Specific child table to return ('maintenance_entries') If None, returns all child tables Returns: Child table data """ try: if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Check if user has permission to read this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'read', pm_schedule_name): frappe.throw(_('Not permitted to access this PM Schedule Generator')) # Get PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) result = {} if child_table: if child_table == 'maintenance_entries': result['maintenance_entries'] = [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] else: frappe.throw(_('Invalid child table name')) else: result = { 'maintenance_entries': [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])] } frappe.response['message'] = { 'success': True, 'pm_schedule_name': pm_schedule_name, **result } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedule Child Tables API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def bulk_update_maintenance_entries(pm_schedule_name, maintenance_entries): """ Bulk update/replace all maintenance entries in a PM Schedule Generator Args: pm_schedule_name: Name/ID of the PM Schedule Generator maintenance_entries: JSON string containing list of maintenance entries Returns: Updated PM Schedule with new maintenance entries """ try: import json if not pm_schedule_name: frappe.throw(_('PM Schedule Generator name is required')) # Parse maintenance entries if isinstance(maintenance_entries, str): maintenance_entries = json.loads(maintenance_entries) # Check if user has permission to update this PM Schedule if not frappe.has_permission('PM Schedule Generator', 'write', pm_schedule_name): frappe.throw(_('Not permitted to update this PM Schedule Generator')) # Get PM Schedule pm_schedule = frappe.get_doc('PM Schedule Generator', pm_schedule_name) # Clear existing maintenance entries and add new ones pm_schedule.set('maintenance_entries', []) for entry in maintenance_entries: pm_schedule.append('maintenance_entries', entry) pm_schedule.save() frappe.db.commit() frappe.response['message'] = { 'success': True, 'maintenance_entries': [item.as_dict() for item in pm_schedule.get('maintenance_entries', [])], 'message': _('Maintenance entries updated successfully') } except Exception as e: frappe.db.rollback() frappe.log_error(frappe.get_traceback(), 'Bulk Update Maintenance Entries API Error') frappe.response['message'] = { 'success': False, 'error': str(e) } @frappe.whitelist(allow_guest=True) def get_pm_schedules_by_hospital(hospital, limit=20, offset=0): """ Get PM Schedules filtered by hospital Args: hospital: Hospital name to filter by limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of PM Schedules for the specified hospital """ try: if not hospital: frappe.throw(_('Hospital name is required')) filters = {'hospital': hospital} # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='creation desc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules By Hospital API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_pm_schedules_by_maintenance_team(maintenance_team, limit=20, offset=0): """ Get PM Schedules filtered by maintenance team Args: maintenance_team: Maintenance team name to filter by limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of PM Schedules for the specified maintenance team """ try: if not maintenance_team: frappe.throw(_('Maintenance team name is required')) filters = {'maintenance_team': maintenance_team} # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='creation desc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules By Maintenance Team API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_pm_schedules_by_assignee(assign_to, limit=20, offset=0): """ Get PM Schedules filtered by assigned user Args: assign_to: Assigned user email to filter by limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of PM Schedules assigned to the specified user """ try: if not assign_to: frappe.throw(_('Assignee email is required')) filters = {'assign_to': assign_to} # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='due_date asc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules By Assignee API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_pm_schedules_by_date_range(start_date=None, end_date=None, limit=20, offset=0): """ Get PM Schedules within a date range Args: start_date: Start date filter (YYYY-MM-DD) end_date: End date filter (YYYY-MM-DD) limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of PM Schedules within the specified date range """ try: filters = {} if start_date: filters['start_date'] = ['>=', start_date] if end_date: filters['end_date'] = ['<=', end_date] # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='start_date asc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules By Date Range API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_pm_schedules_by_periodicity(periodicity, limit=20, offset=0): """ Get PM Schedules filtered by periodicity Args: periodicity: Periodicity to filter by (e.g., 'Monthly', 'Quarterly', 'Yearly') limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of PM Schedules with the specified periodicity """ try: if not periodicity: frappe.throw(_('Periodicity is required')) filters = {'periodicity': periodicity} # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='creation desc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get PM Schedules By Periodicity API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_overdue_pm_schedules(limit=20, offset=0): """ Get PM Schedules that are overdue (due_date < today and docstatus != 2) Args: limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of overdue PM Schedules """ try: from frappe.utils import today filters = { 'due_date': ['<', today()], 'docstatus': ['!=', 2] # Exclude cancelled } # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='due_date asc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get Overdue PM Schedules API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 } @frappe.whitelist(allow_guest=True) def get_upcoming_pm_schedules(days=30, limit=20, offset=0): """ Get PM Schedules due within the specified number of days Args: days: Number of days to look ahead (default: 30) limit: Number of records to return (default: 20) offset: Number of records to skip (default: 0) Returns: List of upcoming PM Schedules """ try: from frappe.utils import today, add_days filters = { 'due_date': ['between', [today(), add_days(today(), int(days))]], 'docstatus': ['!=', 2] # Exclude cancelled } # Get total count total_count = frappe.db.count('PM Schedule Generator', filters=filters) # Get PM schedules pm_schedules = frappe.get_all( 'PM Schedule Generator', filters=filters, fields=PM_SCHEDULE_GENERATOR_FIELDS, limit_page_length=int(limit), limit_start=int(offset), order_by='due_date asc' ) # Include child tables for pm_schedule in pm_schedules: pm_schedule['maintenance_entries'] = get_child_table_data( pm_schedule['name'], 'maintenance_entries', 'PM Entry Line', PM_ENTRY_LINE_FIELDS ) # Calculate has_more has_more = (int(offset) + int(limit)) < total_count frappe.response['message'] = { 'pm_schedules': pm_schedules, 'total_count': total_count, 'limit': int(limit), 'offset': int(offset), 'has_more': has_more } except Exception as e: frappe.log_error(frappe.get_traceback(), 'Get Upcoming PM Schedules API Error') frappe.response['message'] = { 'error': str(e), 'pm_schedules': [], 'total_count': 0 }