import frappe from frappe import _ @frappe.whitelist(allow_guest=True) def get_incidents(): incidents = frappe.get_list( 'SB Incident', filters=_organization_filter(), fields=['name', 'number', 'title', 'organization', 'type', 'status', 'date', 'assigned_to', 'deadline_date'], order_by='creation desc', ignore_permissions=True ) return incidents @frappe.whitelist(allow_guest=True) def create_incident(title, organization, type, description, date, time='', management='', department='', section_field='', branch='', incident_class='', hses_field=''): doc = frappe.new_doc('SB Incident') doc.title = title doc.organization = _user_organization() or organization doc.type = type doc.description = description doc.date = date doc.time = time or '' doc.management = management or '' doc.department = department or '' doc.section_field = section_field or '' doc.branch = branch or '' doc.incident_class = incident_class or '' doc.hses_field = hses_field or '' doc.status = 'New' doc.workflow_state = 'New' # Auto-fill creator data from logged-in user user = frappe.session.user user_doc = frappe.get_doc('User', user) if frappe.db.exists('User', user) else None if user_doc: doc.creator_name = user_doc.full_name or user_doc.first_name or user doc.creator_id = getattr(user_doc, 'employee_id', '') or '' doc.creator_division = getattr(user_doc, 'custom_management', '') or '' doc.creator_department = getattr(user_doc, 'custom_department', '') or '' doc.creator_section = getattr(user_doc, 'custom_section', '') or '' doc.creator_branch = getattr(user_doc, 'custom_branch', '') or '' doc.insert(ignore_permissions=True) frappe.db.commit() return doc.name @frappe.whitelist() def save_incident(name, **kwargs): """Save/update SB Incident fields (Auditor triage + Manager solution).""" doc = frappe.get_doc('SB Incident', name) _enforce_same_organization('SB Incident', doc.organization) if 'assigned_to' in kwargs and not _is_auditor_or_admin(): kwargs.pop('assigned_to', None) allowed = [ 'priority', 'economy_class', 'product_relation', 'repeated', 'entity_scope', 'assigned_to', 'solution', 'cost', 'summary', 'auditor_comment', 'management', 'department', 'section_field', 'branch', 'hses_field', 'incident_class', 'type', 'deadline_date', 'status', ] new_comment = kwargs.get('auditor_comment') comment_changed = new_comment and new_comment != doc.auditor_comment for field in allowed: if field in kwargs and kwargs[field] is not None: setattr(doc, field, kwargs[field]) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() if 'status' in kwargs: _notify_manager_on_status('SB Incident', doc) # Owner/Auditor comment flows to the assigned Manager for acknowledgement (BRD 9.3/13.1) if comment_changed and doc.assigned_to: _send_notification( doc.assigned_to, 'New comment on Incident: {}'.format(doc.title or name), 'SB Incident', name ) return frappe.get_doc('SB Incident', name).as_dict() @frappe.whitelist() def save_idea(name, **kwargs): """Save/update SB Idea fields.""" doc = frappe.get_doc('SB Idea', name) _enforce_same_organization('SB Idea', doc.organization) if 'assigned_to' in kwargs and not _is_auditor_or_admin(): kwargs.pop('assigned_to', None) allowed = [ 'priority', 'economy_class', 'product_relation', 'repeated', 'entity_scope', 'assigned_to', 'solution', 'cost', 'auditor_comment', 'idea_class', 'hses_field', 'management', 'department', 'section_field', 'branch', 'current_condition', 'proposed_condition', 'date', 'status', 'type', 'sb_order', ] new_comment = kwargs.get('auditor_comment') comment_changed = new_comment and new_comment != doc.auditor_comment for field in allowed: if field in kwargs and kwargs[field] is not None: setattr(doc, field, kwargs[field]) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() if 'status' in kwargs: _notify_manager_on_status('SB Idea', doc) # Owner/Auditor comment flows to the assigned Manager for acknowledgement (BRD 13.1) if comment_changed and doc.assigned_to: _send_notification( doc.assigned_to, 'New comment on Idea: {}'.format(doc.title or name), 'SB Idea', name ) return frappe.get_doc('SB Idea', name).as_dict() @frappe.whitelist() def save_order(name, **kwargs): """Save/update SB Order fields.""" doc = frappe.get_doc('SB Order', name) _enforce_same_organization('SB Order', doc.organization) allowed = [ 'priority', 'planned', 'economy_class', 'product_relation', 'entity_scope', 'hses_field', 'auditor_comment', 'management', 'department', 'section_field', 'branch', 'budget', 'actual_cost', 'currency', 'description', 'notes', 'starting_date', 'ending_date', 'type', 'status', 'actor', 'cycle', ] for field in allowed: if field in kwargs and kwargs[field] is not None: setattr(doc, field, kwargs[field]) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() if 'status' in kwargs: _notify_manager_on_status('SB Order', doc) return frappe.get_doc('SB Order', name).as_dict() @frappe.whitelist() def save_process(name, **kwargs): """Save/update SB Process fields.""" doc = frappe.get_doc('SB Process', name) _enforce_same_organization('SB Process', doc.organization) allowed = [ 'chain_location', 'economy_class', 'impact_level', 'product_relation', 'entity_scope', 'hses_field', 'kpi_name', 'kpi_type', 'kpi_measure', 'cost_per_cycle', 'cycle_type', 'long_cycle_date_1', 'long_cycle_date_2', 'long_cycle_date_3', 'long_cycle_date_4', 'management', 'department', 'section_field', 'branch', 'actor', 'cycle', 'date_start', 'date_end', 'type', 'title', 'auditor_comment', ] for field in allowed: if field in kwargs and kwargs[field] is not None: setattr(doc, field, kwargs[field]) doc.flags.ignore_links = True doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() return get_process(name) @frappe.whitelist(allow_guest=True) def get_organizations(): my_org = _user_organization() orgs = frappe.get_list( 'SB Organization', filters={'name': my_org} if my_org else {}, fields=['name', 'organization_name', 'type', 'city', 'is_active'], order_by='organization_name asc', ignore_permissions=True ) return orgs @frappe.whitelist() def get_my_organization(): """The Organization the calling user is restricted to, or '' if unrestricted.""" return _user_organization() or '' @frappe.whitelist(allow_guest=True) def create_organization(organization_name, type, city, contact_email=''): doc = frappe.new_doc('SB Organization') doc.organization_name = organization_name doc.type = type doc.city = city doc.contact_email = contact_email doc.is_active = 1 doc.insert(ignore_permissions=True) frappe.db.commit() return doc.name @frappe.whitelist(allow_guest=True) def update_incident_status(name, status): doc = frappe.get_doc('SB Incident', name) _enforce_same_organization('SB Incident', doc.organization) doc.status = status doc.save(ignore_permissions=True) frappe.db.commit() _notify_manager_on_status('SB Incident', doc) return 'ok' @frappe.whitelist(allow_guest=True) def get_ideas(): ideas = frappe.get_list( 'SB Idea', filters=_organization_filter(), fields=['name', 'number', 'title', 'organization', 'type', 'status', 'sb_order', 'workflow_state', 'assigned_to', 'department', 'branch'], order_by='creation desc', ignore_permissions=True ) return ideas @frappe.whitelist(allow_guest=True) def create_idea(title, organization, category, description, submitted_by, date): doc = frappe.new_doc('SB Idea') doc.title = title doc.organization = _user_organization() or organization doc.insert(ignore_permissions=True) frappe.db.commit() return doc.name @frappe.whitelist(allow_guest=True) def update_idea_status(name, status): doc = frappe.get_doc('SB Idea', name) _enforce_same_organization('SB Idea', doc.organization) doc.status = status doc.save(ignore_permissions=True) frappe.db.commit() _notify_manager_on_status('SB Idea', doc) return 'ok' @frappe.whitelist(allow_guest=True) def get_orders(): orders = frappe.get_list( 'SB Order', filters=_organization_filter(), fields=['name', 'number', 'title', 'organization', 'type', 'status', 'starting_date', 'ending_date', 'stage', 'completion_pct', 'escalation_level', 'sb_idea', 'workflow_state', 'assigned_to', 'actor', 'department', 'branch'], order_by='creation desc', ignore_permissions=True ) return orders @frappe.whitelist(allow_guest=True) def create_order(title, organization, type, description, submitted_by, date): doc = frappe.new_doc('SB Order') doc.title = title doc.organization = _user_organization() or organization doc.type = type doc.description = description doc.stage = 1 doc.completion_pct = 20 doc.status = 'Bidding' doc.workflow_state = 'Draft' doc.flags.ignore_links = True doc.insert(ignore_permissions=True) frappe.db.commit() return doc.name @frappe.whitelist(allow_guest=True) def update_order_status(name, status): doc = frappe.get_doc('SB Order', name) _enforce_same_organization('SB Order', doc.organization) doc.status = status doc.save(ignore_permissions=True) frappe.db.commit() _notify_manager_on_status('SB Order', doc) return 'ok' @frappe.whitelist(allow_guest=True) def get_processes(): processes = frappe.get_list('SB Process', filters=_organization_filter(), fields=['name', 'number', 'title', 'type', 'status', 'cycle_type', 'date_start', 'date_end', 'organization', 'escalation_level', 'actor', 'cycle'], order_by='creation desc', ignore_permissions=True) return processes @frappe.whitelist(allow_guest=True) def get_users(): filters = {'enabled': 1, 'name': ['not in', ['Guest']]} from sidrah_business.soft_delete import ( is_platform_admin, get_user_group_company, is_ceo_user, is_group_company_admin, is_manager_user, is_auditor_user, get_subordinates_recursive ) user = frappe.session.user if user and user not in ('Guest', 'Administrator'): my_co = get_user_group_company(user) # CEO without company has global visibility; all others are scoped to their group company if not (is_ceo_user(user) and not my_co): if my_co: filters['custom_group_company'] = my_co users = frappe.get_all('User', filters=filters, fields=['name', 'full_name', 'email', 'custom_organization', 'custom_group_company', 'custom_manager', 'custom_category', 'enabled'], ignore_permissions=True ) if not user or user in ('Guest', 'Administrator') or is_ceo_user(user) or is_group_company_admin(user): return users # Manager: sees self + all users reporting under him (auditors + employees) + his direct manager if is_manager_user(user): subs = set(get_subordinates_recursive(user)) allowed = subs | {user} mgr = frappe.db.get_value('User', user, 'custom_manager') if mgr: allowed.add(mgr) return [u for u in users if u.get('name') in allowed] # Auditor: sees self + all employees reporting under him + his manager if is_auditor_user(user): subs = set(get_subordinates_recursive(user)) allowed = subs | {user} mgr = frappe.db.get_value('User', user, 'custom_manager') if mgr: allowed.add(mgr) return [u for u in users if u.get('name') in allowed] # Employee: sees self + his direct manager/auditor allowed = {user} mgr = frappe.db.get_value('User', user, 'custom_manager') if mgr: allowed.add(mgr) return [u for u in users if u.get('name') in allowed] @frappe.whitelist(allow_guest=True) def assign_incident(name, assigned_to): doc = frappe.get_doc('SB Incident', name) _enforce_same_organization('SB Incident', doc.organization) doc.assigned_to = assigned_to doc.save(ignore_permissions=True) frappe.db.commit() return 'ok' @frappe.whitelist(allow_guest=True) def get_user_roles(user): roles = frappe.get_all('Has Role', filters={'parent': user}, fields=['role'], ignore_permissions=True ) role_list = [r.role for r in roles] cat = frappe.db.get_value('User', user, 'custom_category') if cat: if cat not in role_list: role_list.append(cat) cat_map = { 'Admin': ['SB Admin', 'System Manager'], 'Business Owner': ['SB Business Owner', 'Business Owner', 'SB Owner'], 'Manager': ['SB Manager'], 'Auditor': ['SB Auditor'], 'X-Auditor': ['SB X-Auditor', 'SB Auditor'], 'Employee': ['SB Employee'], 'CEO': ['Ceo', 'SB Admin', 'System Manager'], } for r in cat_map.get(cat, []): if r not in role_list: role_list.append(r) return role_list @frappe.whitelist() def get_user_role_map(): """Map of user -> [SB roles] for all users. Bypasses the 'Has Role' doctype's System Manager-only read permission so any logged-in user (e.g. an Auditor building an Actor/Manager dropdown) can resolve role membership.""" sb_roles = [ 'SB Manager', 'SB Admin', 'SB Owner', 'SB Auditor', 'SB X-Auditor', 'SB Employee', 'System Manager', 'Business Owner', 'SB Business Owner', 'Ceo', 'CEO', 'SB CEO' ] rows = frappe.get_all('Has Role', filters={'role': ['in', sb_roles], 'parenttype': 'User'}, fields=['parent', 'role'], ignore_permissions=True ) result = {} for r in rows: result.setdefault(r.parent, []).append(r.role) # Fallback to custom_category so newly created users immediately appear in role-based pickers category_map = { 'Admin': ['SB Admin', 'System Manager'], 'Business Owner': ['SB Business Owner', 'Business Owner', 'SB Owner'], 'Manager': ['SB Manager'], 'Auditor': ['SB Auditor'], 'X-Auditor': ['SB X-Auditor', 'SB Auditor'], 'Employee': ['SB Employee'], 'CEO': ['Ceo', 'SB Admin', 'System Manager'], } users_with_cat = frappe.get_all('User', filters={'custom_category': ['is', 'set'], 'enabled': 1}, fields=['name', 'custom_category'], ignore_permissions=True ) for u in users_with_cat: cat_roles = category_map.get(u.custom_category, []) for r in cat_roles: if r not in result.setdefault(u.name, []): result[u.name].append(r) return result @frappe.whitelist() def set_user_password(user, password): """Safely updates user password in __Auth without sending emails or throwing SMTP errors.""" from sidrah_business.soft_delete import is_admin_user, is_business_owner_user if not (is_admin_user() or is_business_owner_user() or frappe.session.user == user): frappe.throw(_('Permission denied to update password.'), frappe.PermissionError) from frappe.utils.password import update_password update_password(user, password) return {'success': True} @frappe.whitelist() def get_csrf_token(): return frappe.sessions.get_csrf_token() # ── Process Workflow ────────────────────────────────────────────────────────── _WORKFLOW_TRANSITIONS = { 'Draft': {'Submit for Review': 'Pending Actor'}, 'Pending Actor': {'Update Status': 'Pending Comment'}, 'Pending Comment': {'Add Comment': 'Review'}, 'Review': { 'Move to Study': 'In Study', 'Start New Cycle': 'Pending Actor', 'Close Process': 'Closed', }, 'In Study': { 'Start New Cycle': 'Pending Actor', 'Close Process': 'Closed', }, } @frappe.whitelist() def get_process(name): """Return SB Process doc bypassing permlevel restrictions (for Actor view).""" doc = frappe.get_doc('SB Process', name) _enforce_same_organization('SB Process', doc.organization) doc.flags.ignore_permissions = True return doc.as_dict() @frappe.whitelist() def apply_process_workflow(name, action, actor_status=None, actor_comment=None, progress_indicators=None): """Advance workflow_state for SB Process. Actor manually selects Pass/Fail status.""" import json as _json from frappe.utils import nowdate from sidrah_business.tasks import _apply_process_escalation doc = frappe.get_doc('SB Process', name) _enforce_same_organization('SB Process', doc.organization) current_state = doc.workflow_state or 'Draft' state_transitions = _WORKFLOW_TRANSITIONS.get(current_state, {}) if action not in state_transitions: valid = list(state_transitions.keys()) or ['none'] frappe.throw( "Action '{}' is not valid in state '{}'. Valid actions: {}".format( action, current_state, ', '.join(valid) ) ) _enforce_process_action_permission(action) if action == 'Start New Cycle': if getattr(doc, 'completion_pct', None) == 100 or getattr(doc, 'stage', None) == 5 or getattr(doc, 'workflow_state', None) in ('Completed', 'Closed'): frappe.throw('This Process is 100% completed. No further cycles are required.') if action == 'Update Status': roles = frappe.get_roles(frappe.session.user) is_privileged = any(r in roles for r in ('SB Admin', 'SB Auditor', 'SB Owner', 'System Manager')) if 'SB Manager' in roles and not is_privileged: if doc.actor != frappe.session.user: frappe.throw('You are not the assigned Actor for this Process.') # Rule 11.1-1: Notify Actor when Auditor submits for review if action == 'Submit for Review': actor = getattr(doc, 'actor', None) if not actor: frappe.throw('Select an Actor before submitting for review — otherwise no one will see this in their Inbox.') _send_notification( actor, 'Process assigned to you — please update status before cycle ends: {}'.format(doc.title or name), 'SB Process', name ) # Actor enters the Actual Value per indicator; auto-compute Normal/Below Min/Above Max if action == 'Update Status' and progress_indicators: try: indicator_updates = _json.loads(progress_indicators) if isinstance(progress_indicators, str) else progress_indicators except (ValueError, TypeError): indicator_updates = [] updates_by_name = {u.get('name'): u for u in indicator_updates if u.get('name')} for row in (doc.progress_indicators or []): upd = updates_by_name.get(row.name) if not upd: continue if upd.get('actual_value') not in (None, ''): try: row.actual_value = float(upd['actual_value']) except (TypeError, ValueError): pass if upd.get('cycle_comment') is not None: row.cycle_comment = upd['cycle_comment'] row.cycle_date = nowdate() row.cycle = doc.cycle or row.cycle if row.actual_value is not None: try: mn = float(row.minimum) if row.minimum not in (None, '') else None except (TypeError, ValueError): mn = None try: mx = float(row.maximum) if row.maximum not in (None, '') else None except (TypeError, ValueError): mx = None if mn is not None and row.actual_value < mn: row.status = 'Below Min' elif mx is not None and row.actual_value > mx: row.status = 'Above Max' else: row.status = 'Normal' row.cycle_status = row.status # Rule 11.1-1/2/5: Actor manually sets Pass/Fail; log cycle entry; re-evaluate escalation if action == 'Update Status': manual_status = actor_status if actor_status in ('Pass', 'Fail') else 'Unknown' # BRD 4.4/7.1/11.4 — a 'Long' cycle_type process reports against its own # milestone dates, not the org's regular Cycle, so it's deduplicated by # milestone date instead of by `doc.cycle`. if doc.cycle_type == 'Long': from sidrah_business.tasks import _current_long_cycle_milestone milestone = _current_long_cycle_milestone(doc) period_key = str(milestone) if milestone else nowdate() existing = [r for r in (doc.cycle_logs or []) if str(r.cycle_date) == period_key] else: period_key = doc.cycle or '' existing = [r for r in (doc.cycle_logs or []) if r.cycle == period_key] if existing: # One log row per cycle/milestone — a re-submit (or a correction after the # engine auto-logged Unknown) must overwrite that row, not be silently # dropped, or escalation keeps evaluating the stale status forever. log_row = existing[-1] else: log_row = doc.append('cycle_logs', {}) log_row.cycle = doc.cycle or '' log_row.cycle_date = period_key if doc.cycle_type == 'Long' else nowdate() log_row.status = manual_status log_row.actor = frappe.session.user if actor_comment: log_row.comment = actor_comment doc.status = manual_status _apply_process_escalation(doc) # Actor submitted their cycle status — Auditor/Manager now need to add a # comment before it can move to Review (unconditional hand-off, same reasoning # as the Idea "Submit Proposal" notice above). _notify_org_auditors( doc.organization, 'Process needs a comment before review: {}'.format(doc.title or name), 'SB Process', name, exclude=frappe.session.user ) _notify_org_managers( doc.organization, 'Process needs a comment before review: {}'.format(doc.title or name), 'SB Process', name, exclude=frappe.session.user ) # Add Comment moves the process to Review — notify the Actor about the comment # (BRD 13.1 "Actor + Manager"), and Auditor/Manager broadly that a Move to Study / # Start New Cycle decision is now needed, in one notification per audience. if action == 'Add Comment': if actor_comment: doc.auditor_comment = actor_comment if doc.actor: _send_notification( doc.actor, 'New comment on process: {}'.format(doc.title or name), 'SB Process', name ) subject = 'Process ready for review decision (Move to Study / New Cycle): {}'.format(doc.title or name) _notify_org_managers(doc.organization, subject, 'SB Process', name, exclude=doc.actor) _notify_org_auditors(doc.organization, subject, 'SB Process', name, exclude=doc.actor) # Moved to Study — Auditor/Manager need to eventually Close or Start New Cycle if action == 'Move to Study': _notify_org_auditors( doc.organization, 'Process in Study — close or start a new cycle when ready: {}'.format(doc.title or name), 'SB Process', name, exclude=frappe.session.user ) _notify_org_managers( doc.organization, 'Process in Study — close or start a new cycle when ready: {}'.format(doc.title or name), 'SB Process', name, exclude=frappe.session.user ) doc.flags.ignore_links = True doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() if action == 'Update Status': _notify_manager_on_status('SB Process', doc) next_state = state_transitions[action] frappe.db.set_value('SB Process', name, 'workflow_state', next_state, update_modified=False) frappe.db.commit() _log_version('SB Process', name, [['workflow_state', current_state, next_state]]) doc.workflow_state = next_state _fire_state_alert('SB Process', doc) # Close Process — clear escalation so it leaves the active dashboard and set 100% completion if action == 'Close Process': doc.completion_pct = 100 doc.status = 'Completed' frappe.db.set_value('SB Process', name, 'escalation_level', 0, update_modified=False) frappe.db.commit() # Notify actor on any transition back to Pending Actor (e.g. Start New Cycle) if next_state == 'Pending Actor' and action != 'Submit for Review': actor = getattr(doc, 'actor', None) if actor: _send_notification( actor, 'New cycle started — update status: {}'.format(doc.title or name), 'SB Process', name ) return get_process(name) # ── Order Workflow ──────────────────────────────────────────────────────────── _STAGE_TO_STATUS = {1: 'Bidding', 2: 'Execution', 3: 'Validation', 4: 'Payment', 5: 'Completion'} _ORDER_WORKFLOW_TRANSITIONS = { 'Draft': {'Submit for Review': 'Pending Actor'}, 'Pending Actor': {'Update Status': 'Review'}, 'Review': { 'Move to Study': 'In Study', 'Start New Cycle': 'Pending Actor', 'Close Order': 'Closed', }, 'In Study': { 'Start New Cycle': 'Pending Actor', 'Close Order': 'Closed', }, } @frappe.whitelist() def apply_order_workflow(name, action, comment='', stage=None, status=None, cycle_status=None): """Advance outsourcing order through 4-step workflow and log the cycle entry.""" from frappe.utils import nowdate doc = frappe.get_doc('SB Order', name) _enforce_same_organization('SB Order', doc.organization) current_state = doc.workflow_state or 'Draft' state_transitions = _ORDER_WORKFLOW_TRANSITIONS.get(current_state, {}) if action not in state_transitions: valid = list(state_transitions.keys()) or ['none'] frappe.throw( "Action '{}' is not valid in state '{}'. Valid actions: {}".format( action, current_state, ', '.join(valid) ) ) _enforce_order_action_permission(doc, action) if action == 'Start New Cycle': if getattr(doc, 'stage', None) == 5 or getattr(doc, 'completion_pct', None) == 100 or getattr(doc, 'workflow_state', None) in ('Completed', 'Closed'): last_status = cycle_status if not last_status and getattr(doc, 'order_cycle_log', None) and len(doc.order_cycle_log) > 0: last_status = doc.order_cycle_log[-1].status if last_status not in ('Deviate', 'Unknown'): frappe.throw('This Order is at Stage 5 (100% Completion). No further cycles are required.') next_state = state_transitions[action] if stage is not None: try: stage_int = int(stage) doc.stage = stage_int doc.completion_pct = stage_int * 20 doc.status = _STAGE_TO_STATUS.get(stage_int, 'Bidding') except (ValueError, TypeError): pass if status: doc.status = status if action == 'Update Status': # cycle_status must be Agree / Deviate / Unknown; default Agree from sidrah_business.tasks import _apply_order_escalation valid_cycle_statuses = ('Agree', 'Deviate', 'Unknown') resolved_cycle_status = cycle_status if cycle_status in valid_cycle_statuses else 'Agree' log_row = doc.append('order_cycle_log', {}) log_row.cycle = doc.cycle log_row.cycle_date = nowdate() log_row.stage = doc.stage or 1 log_row.status = resolved_cycle_status log_row.actor = frappe.session.user log_row.comment = comment # Rule 11.2-1/2/5: re-evaluate escalation after every actor submit _apply_order_escalation(doc) if action == 'Close Order': doc.stage = 5 doc.completion_pct = 100 doc.status = 'Closed' frappe.db.set_value('SB Order', name, 'escalation_level', 0, update_modified=False) if comment: doc.auditor_comment = comment # Owner/Auditor comment flows to Actor + Manager for acknowledgement (BRD 8.3/13.1) — # not on the Actor's own Update Status submission, which is their comment, not one *to* them if action != 'Update Status': actor_recipient = getattr(doc, 'actor', None) or getattr(doc, 'assigned_to', None) subject = 'New comment on outsourcing order: {}'.format(doc.title or name) if actor_recipient: _send_notification(actor_recipient, subject, 'SB Order', name) _notify_org_managers(doc.organization, subject, 'SB Order', name, exclude=actor_recipient) # Unconditional hand-off notices — whoever needs to act next on this stage # (Auditor/Manager/Owner, per _enforce_order_action_permission) hears about it, # same reasoning as the Idea/Process ones above. if next_state == 'Review': subject = 'Order ready for review decision (Move to Study / New Cycle): {}'.format(doc.title or name) _notify_org_managers(doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) _notify_org_auditors(doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) _notify_org_role('SB Owner', doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) elif next_state == 'In Study': subject = 'Order in Study — close or start a new cycle when ready: {}'.format(doc.title or name) _notify_org_managers(doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) _notify_org_auditors(doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) _notify_org_role('SB Owner', doc.organization, subject, 'SB Order', name, exclude=frappe.session.user) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() if stage is not None or status or action == 'Close Order': _notify_manager_on_status('SB Order', doc) frappe.db.set_value('SB Order', name, 'workflow_state', next_state, update_modified=False) frappe.db.commit() _log_version('SB Order', name, [['workflow_state', current_state, next_state]]) doc.workflow_state = next_state _fire_state_alert('SB Order', doc) # Notify actor when order moves to Pending Actor if next_state == 'Pending Actor': actor_user = getattr(doc, 'actor', None) or getattr(doc, 'assigned_to', None) if actor_user: _send_notification( actor_user, 'Outsourcing order requires your status update: {}'.format(doc.title or name), 'SB Order', name ) fresh = frappe.get_doc('SB Order', name) return fresh.as_dict() # ── Incident Workflow ───────────────────────────────────────────────────────── _INCIDENT_WORKFLOW_TRANSITIONS = { 'New': {'Submit': 'Initial Evaluation'}, 'Initial Evaluation': { 'Assign to Manager': 'Accept & Under Progress', 'Reject': 'Rejected', }, 'Accept & Under Progress': { 'Submit Solution': 'working on resolution', }, 'working on resolution': { 'Resolution complete': 'Executed & Closed', }, } _INCIDENT_STATE_STATUS = { 'Initial Evaluation': 'Initial Evaluation', 'Accept & Under Progress': 'Accept & Under Progress', 'working on resolution': 'Accept & Under Progress', 'Executed & Closed': 'Executed & Closed', 'Rejected': 'Rejected', } @frappe.whitelist() def apply_incident_workflow(name, action, fields_json=None): """Progress events/incidents through their 5-step lifecycle.""" import json as _json doc = frappe.get_doc('SB Incident', name) _enforce_same_organization('SB Incident', doc.organization) extra = {} if fields_json: extra = _json.loads(fields_json) if isinstance(fields_json, str) else fields_json current_state = doc.workflow_state or 'New' state_transitions = _INCIDENT_WORKFLOW_TRANSITIONS.get(current_state, {}) if action not in state_transitions: valid = list(state_transitions.keys()) or ['none'] frappe.throw( "Action '{}' is not valid in state '{}'. Valid actions: {}".format( action, current_state, ', '.join(valid) ) ) _enforce_incident_idea_action_permission(action, doctype='SB Incident', doc=doc) next_state = state_transitions[action] for k, v in extra.items(): if hasattr(doc, k): setattr(doc, k, v) if next_state in _INCIDENT_STATE_STATUS: doc.status = _INCIDENT_STATE_STATUS[next_state] elif action == 'Reject': doc.status = 'Rejected' # Rule: only page the Manager/Actor if the new status's master record has # 'Notify Manager' checked and is itself Active (see _notify_manager_on_status) — # this is the extra, configurable layer on top of the unconditional notice below. _notify_manager_on_status('SB Incident', doc) # Rule 11.3-1: Notify Auditors when incident enters the queue if action == 'Submit': auditors = frappe.get_all('Has Role', filters={'role': ['in', ['SB Auditor', 'SB Admin', 'System Manager']]}, fields=['parent'], ignore_permissions=True ) seen = set() for a in auditors: if a.parent not in seen: _send_notification( a.parent, 'New Incident requires triage: {}'.format(doc.title or name), 'SB Incident', name ) seen.add(a.parent) # Assign to Manager — unconditional "you have a new assignment" notice (BRD 13.1 # "Event assigned to Manager"), independent of the configurable notify_manager flag above. if action == 'Assign to Manager' and doc.assigned_to: _send_notification( doc.assigned_to, 'Incident assigned to you: {}'.format(doc.title or name), 'SB Incident', name ) # Reject — let the original creator know their report was not accepted. if action == 'Reject' and doc.owner: _send_notification( doc.owner, 'Your reported incident {} was rejected at triage'.format(doc.title or name), 'SB Incident', name ) doc.flags.ignore_links = True doc.flags.ignore_workflow = True doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() frappe.db.set_value('SB Incident', name, 'workflow_state', next_state, update_modified=False) frappe.db.commit() _log_version('SB Incident', name, [['workflow_state', current_state, next_state]]) doc.workflow_state = next_state _fire_state_alert('SB Incident', doc) return frappe.get_doc('SB Incident', name).as_dict() # ── Idea / Study Workflow ───────────────────────────────────────────────────── _IDEA_WORKFLOW_TRANSITIONS = { 'New': {'Submit': 'Initial Evaluation'}, 'Initial Evaluation': { 'Assign to Manager': 'Accept & Under Progress', 'Reject': 'Rejected', }, 'Accept & Under Progress': { 'Submit Proposal': 'working on proposal', }, 'working on proposal': { 'Move to Outsourcing': 'Executed & Closed', 'Reject': 'Rejected', }, } _IDEA_STATE_STATUS = { 'Initial Evaluation': 'Initial Evaluation', 'Accept & Under Progress': 'Accept & Under Progress', 'working on proposal': 'Accept & Under Progress', 'Executed & Closed': 'Executed & Closed', 'Rejected': 'Rejected', } @frappe.whitelist() def apply_idea_workflow(name, action, fields_json=None): """Progress study/ideas through their 5-step lifecycle, ending with Move to Outsourcing.""" import json as _json doc = frappe.get_doc('SB Idea', name) _enforce_same_organization('SB Idea', doc.organization) extra = {} if fields_json: extra = _json.loads(fields_json) if isinstance(fields_json, str) else fields_json current_state = doc.workflow_state or 'New' state_transitions = _IDEA_WORKFLOW_TRANSITIONS.get(current_state, {}) if action not in state_transitions: valid = list(state_transitions.keys()) or ['none'] frappe.throw( "Action '{}' is not valid in state '{}'. Valid actions: {}".format( action, current_state, ', '.join(valid) ) ) _enforce_incident_idea_action_permission(action, doctype='SB Idea', doc=doc) next_state = state_transitions[action] for k, v in extra.items(): if hasattr(doc, k): setattr(doc, k, v) created_order_name = None created_order_number = None if action == 'Move to Outsourcing': # Create an SB Order linked to this Idea order_doc = frappe.new_doc('SB Order') order_doc.title = doc.title or 'Outsourced Idea' order_doc.organization = doc.organization order_doc.management = doc.management or '' order_doc.department = doc.department or '' order_doc.section_field = doc.section_field or '' order_doc.branch = doc.branch or '' order_doc.priority = doc.priority or 'Normal' order_doc.economy_class = doc.economy_class or '' order_doc.entity_scope = doc.entity_scope or '' order_doc.product_relation = doc.product_relation or 0 order_doc.hses_field = doc.hses_field or '' # Build description from Problem / Current condition, Proposed condition, and Solution desc_parts = [] if doc.current_condition: desc_parts.append("Current Status:\n" + doc.current_condition) if doc.proposed_condition: desc_parts.append("Proposed Status:\n" + doc.proposed_condition) if doc.solution: desc_parts.append("Solution/Plan:\n" + doc.solution) order_doc.description = "\n\n".join(desc_parts) if desc_parts else (doc.title or '') if doc.cost: order_doc.budget = doc.cost order_doc.actual_cost = doc.cost order_doc.starting_date = frappe.utils.today() order_doc.status = 'New' order_doc.workflow_state = 'Draft' order_doc.stage = 1 order_doc.completion_pct = 20 order_doc.sb_idea = doc.name # Match order type if possible if doc.type and frappe.db.exists('Order Types', doc.type): order_doc.type = doc.type else: first_type = frappe.db.get_value('Order Types', {}, 'name') if first_type: order_doc.type = first_type order_doc.flags.ignore_permissions = True order_doc.flags.ignore_links = True order_doc.insert(ignore_permissions=True) created_order_name = order_doc.name created_order_number = order_doc.number doc.sb_order = order_doc.name next_state = 'Executed & Closed' if next_state in _IDEA_STATE_STATUS: doc.status = _IDEA_STATE_STATUS[next_state] # Rule: only page the Manager/Actor if the new status's master record has # 'Notify Manager' checked and is itself Active (see _notify_manager_on_status). _notify_manager_on_status('SB Idea', doc) # Notify Auditors when an idea/study enters the queue (mirrors Incident's Submit) if action == 'Submit': _notify_org_auditors( doc.organization, 'New Idea requires evaluation: {}'.format(doc.title or name), 'SB Idea', name ) # Manager submitted their proposal — this is the last step before the Auditor's # final call (Move to Outsourcing / Reject), so the Auditor needs to hear about it # unconditionally, same as the initial "Submit" triage notice — not gated by the # configurable notify_manager flag, which is a different, Manager-facing setting. if action == 'Submit Proposal': _notify_org_auditors( doc.organization, 'Idea ready for your final decision: {}'.format(doc.title or name), 'SB Idea', name ) # Assign to Manager — unconditional "you have a new assignment" notice, independent # of the configurable notify_manager flag above. if action == 'Assign to Manager' and getattr(doc, 'assigned_to', None): _send_notification( doc.assigned_to, 'Idea assigned to you: {}'.format(doc.title or name), 'SB Idea', name ) # Reject (from triage or from the Auditor's final decision) — let the original # creator know their submission was not accepted. if action == 'Reject' and doc.owner: _send_notification( doc.owner, 'Your submitted idea {} was rejected'.format(doc.title or name), 'SB Idea', name ) # Move to Outsourcing — notification of conversion to Order if action == 'Move to Outsourcing' and created_order_name: _notify_org_auditors( doc.organization, 'Idea {} was converted to Order {}'.format(doc.title or doc.number or name, created_order_number or created_order_name), 'SB Order', created_order_name ) doc.flags.ignore_links = True doc.flags.ignore_workflow = True doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() frappe.db.set_value('SB Idea', name, 'workflow_state', next_state, update_modified=False) if created_order_name: frappe.db.set_value('SB Idea', name, 'sb_order', created_order_name, update_modified=False) frappe.db.commit() _log_version('SB Idea', name, [['workflow_state', current_state, next_state]]) doc.workflow_state = next_state _fire_state_alert('SB Idea', doc) res = frappe.get_doc('SB Idea', name).as_dict() if created_order_name: res['created_order'] = created_order_name res['order_number'] = created_order_number return res # ── Dashboard Data ──────────────────────────────────────────────────────────── @frappe.whitelist() def get_dashboard_data(organization=None): """Return aggregated counts + escalated/attention items for dashboard.""" from datetime import date, timedelta organization = _user_organization() or organization filters = {} if organization: # A record with no organization set isn't owned by any one org, so it stays # visible regardless of which org the dashboard is filtered to. filters['organization'] = ['in', [organization, '', None]] fields_proc = ['name', 'number', 'title', 'organization', 'status', 'escalation_level'] fields_order = ['name', 'number', 'title', 'organization', 'status', 'escalation_level'] fields_inc = ['name', 'number', 'title', 'organization', 'date'] # Level 2 — critical escalations (Rules 11.1-4, 11.2-4); exclude closed processes escalated_processes = frappe.get_list('SB Process', filters={**filters, 'escalation_level': ['>=', 2], 'workflow_state': ['!=', 'Closed']}, fields=fields_proc, ignore_permissions=True ) escalated_orders = frappe.get_list('SB Order', filters={**filters, 'escalation_level': ['>=', 2], 'workflow_state': ['!=', 'Closed']}, fields=fields_order, ignore_permissions=True ) # Level 1 — attention needed (single Fail or Deviate, Rules 11.1-2, 11.2-2); exclude closed attention_processes = frappe.get_list('SB Process', filters={**filters, 'escalation_level': 1, 'workflow_state': ['!=', 'Closed']}, fields=fields_proc, ignore_permissions=True ) attention_orders = frappe.get_list('SB Order', filters={**filters, 'escalation_level': 1, 'workflow_state': ['!=', 'Closed']}, fields=fields_order, ignore_permissions=True ) # Untriaged incidents (still New) — Rule 11.3-1 untriaged_incidents = frappe.get_list('SB Incident', filters={**filters, 'workflow_state': 'New'}, fields=fields_inc, ignore_permissions=True ) # Overdue triage (New + past 3-day window) — Rule 11.3-3 triage_cutoff = str(date.today() - timedelta(days=3)) overdue_triage = frappe.get_list('SB Incident', filters={**filters, 'workflow_state': 'New', 'creation': ['<', triage_cutoff]}, fields=fields_inc, ignore_permissions=True ) kpi_filters = {} if organization: kpi_filters['organization'] = ['in', [organization, '', None]] kpi_inputs = frappe.get_list('SB KPI Input', filters=kpi_filters, fields=['name', 'organization', 'period', 'revenue_growth', 'net_profit', 'cash_flow', 'strategic_goal_pct', 'customer_satisfaction', 'employee_turnover_pct', 'open_safety_incidents', 'compliance_score'], order_by='modified desc', limit=5, ignore_permissions=True ) # Cycle-specific Pass/Fail/Unknown counts for current active cycle (Level 1 dashboard) cycle_filter = {'status': 'Active'} if organization: cycle_filter['organization'] = ['in', [organization, '', None]] active_cycles = frappe.get_list('SB Cycle', filters=cycle_filter, fields=['name'], limit=1, order_by='end_date desc', ignore_permissions=True ) cycle_stats = {'pass': 0, 'fail': 0, 'unknown': 0, 'cycle': ''} if active_cycles: _cyc = active_cycles[0].name cycle_stats['cycle'] = _cyc for s in frappe.db.sql( "SELECT status, COUNT(*) AS cnt FROM `tabSB Process Cycle Log` WHERE cycle=%s GROUP BY status", _cyc, as_dict=True ): if s.status == 'Pass': cycle_stats['pass'] = s.cnt elif s.status == 'Fail': cycle_stats['fail'] = s.cnt elif s.status == 'Unknown': cycle_stats['unknown'] = s.cnt return { 'escalated_processes': escalated_processes, 'escalated_orders': escalated_orders, 'attention_processes': attention_processes, 'attention_orders': attention_orders, 'untriaged_incidents': untriaged_incidents, 'overdue_triage': overdue_triage, 'kpi_inputs': kpi_inputs, 'cycle_stats': cycle_stats, } # ── Reports ─────────────────────────────────────────────────────────────────── @frappe.whitelist() def get_report_data(report_name, filters_json=None): """Return rows and columns for a named report.""" import json as _json f = {} if filters_json: f = _json.loads(filters_json) if isinstance(filters_json, str) else filters_json org = _user_organization() or f.get('organization') dept = f.get('department') from_date = f.get('from_date') to_date = f.get('to_date') status = f.get('status') def base_filters(doctype_org_field='organization', date_field=None): filters = {} if org: filters[doctype_org_field] = ['in', [org, '', None]] if dept: filters['department'] = dept if status: filters['status'] = status # date_field names which column From/To constrain — each report picks the one # that's actually shown as a column, so filtering out a row is never a mystery. # The upper bound always gets an end-of-day time: harmless for plain Date columns # (MySQL widens them to midnight either way) and required for Datetime ones # (e.g. creation) so a To date doesn't cut off everything after 00:00 that day. if date_field and (from_date or to_date): to_bound = (to_date + ' 23:59:59') if to_date else None if from_date and to_date: filters[date_field] = ['between', [from_date, to_bound]] elif from_date: filters[date_field] = ['>=', from_date] else: filters[date_field] = ['<=', to_bound] return filters if report_name == 'process_status_history': rows = frappe.get_list('SB Process', filters=base_filters(date_field='date_start'), fields=['name', 'number', 'title', 'organization', 'department', 'status', 'actor', 'cycle', 'date_start', 'date_end', 'escalation_level'], order_by='creation desc', ignore_permissions=True) columns = ['number', 'title', 'organization', 'department', 'status', 'actor', 'cycle', 'date_start', 'date_end', 'escalation_level'] elif report_name == 'order_tracker': rows = frappe.get_list('SB Order', filters=base_filters(date_field='starting_date'), fields=['name', 'number', 'title', 'organization', 'department', 'status', 'stage', 'completion_pct', 'budget', 'starting_date', 'ending_date', 'escalation_level'], order_by='creation desc', ignore_permissions=True) columns = ['number', 'title', 'organization', 'status', 'stage', 'completion_pct', 'budget', 'starting_date', 'ending_date'] elif report_name == 'incident_register': rows = frappe.get_list('SB Incident', filters=base_filters(date_field='date'), fields=['name', 'number', 'title', 'organization', 'department', 'type', 'status', 'date', 'assigned_to'], order_by='date desc', ignore_permissions=True) columns = ['number', 'title', 'organization', 'type', 'status', 'date', 'assigned_to'] elif report_name == 'ideas_pipeline': rows = frappe.get_list('SB Idea', filters=base_filters(date_field='date'), fields=['name', 'number', 'title', 'organization', 'department', 'type', 'status', 'date'], order_by='creation desc', ignore_permissions=True) columns = ['number', 'title', 'organization', 'type', 'status', 'date'] elif report_name == 'department_performance': conditions = ['department IS NOT NULL'] values = {} if org: conditions.append("(organization = %(org)s OR organization IS NULL OR organization = '')") values['org'] = org if from_date and to_date: conditions.append('creation BETWEEN %(from_date)s AND %(to_date)s') values['from_date'] = from_date values['to_date'] = to_date + ' 23:59:59' elif from_date: conditions.append('creation >= %(from_date)s') values['from_date'] = from_date elif to_date: conditions.append('creation <= %(to_date)s') values['to_date'] = to_date + ' 23:59:59' rows = frappe.db.sql(""" SELECT department, SUM(CASE WHEN status = 'Pass' THEN 1 ELSE 0 END) AS pass_count, SUM(CASE WHEN status = 'Fail' THEN 1 ELSE 0 END) AS fail_count, SUM(CASE WHEN status = 'Unknown' THEN 1 ELSE 0 END) AS unknown_count, COUNT(*) AS total FROM `tabSB Process` WHERE {conditions} GROUP BY department ORDER BY department """.format(conditions=' AND '.join(conditions)), values, as_dict=True) columns = ['department', 'pass_count', 'fail_count', 'unknown_count', 'total'] elif report_name == 'process_year_end': from datetime import date as _date year = int(f.get('year', _date.today().year)) year_start = '{}-01-01'.format(year) year_end = '{}-12-31'.format(year) org_filter = '' values = {'year_start': year_start, 'year_end': year_end} if org: org_filter = 'AND p.organization = %(org)s' values['org'] = org rows = frappe.db.sql(""" SELECT p.name, p.number, p.title, p.organization, p.department, p.actor, p.cycle_type, p.cost_per_cycle, COUNT(cl.name) AS total_cycles, SUM(CASE WHEN cl.status = 'Pass' THEN 1 ELSE 0 END) AS pass_count, SUM(CASE WHEN cl.status = 'Fail' THEN 1 ELSE 0 END) AS fail_count, SUM(CASE WHEN cl.status = 'Unknown' THEN 1 ELSE 0 END) AS unknown_count, COALESCE(p.cost_per_cycle, 0) * COUNT(cl.name) AS total_cost FROM `tabSB Process` p LEFT JOIN `tabSB Process Cycle Log` cl ON cl.parent = p.name AND cl.cycle_date BETWEEN %(year_start)s AND %(year_end)s WHERE 1=1 {org_filter} GROUP BY p.name ORDER BY total_cost DESC """.format(org_filter=org_filter), values=values, as_dict=True) columns = [ 'number', 'title', 'organization', 'department', 'actor', 'cycle_type', 'total_cycles', 'pass_count', 'fail_count', 'unknown_count', 'cost_per_cycle', 'total_cost', ] elif report_name == 'escalation_report': esc_filters = base_filters(date_field='creation') esc_filters['escalation_level'] = ['>=', 2] proc = frappe.get_list('SB Process', filters=esc_filters, fields=['name', 'number', 'title', 'organization', 'status', 'escalation_level'], ignore_permissions=True) ords = frappe.get_list('SB Order', filters=esc_filters, fields=['name', 'number', 'title', 'organization', 'status', 'escalation_level'], ignore_permissions=True) rows = [{'entity_type': 'Process', **r} for r in proc] + [{'entity_type': 'Order', **r} for r in ords] columns = ['entity_type', 'number', 'title', 'organization', 'status', 'escalation_level'] elif report_name == 'cost_summary': rows = frappe.get_list('SB Order', filters=base_filters(date_field='starting_date'), fields=['name', 'number', 'title', 'organization', 'department', 'budget', 'actual_cost', 'status', 'starting_date'], order_by='budget desc', ignore_permissions=True) for r in rows: budget = r.get('budget') or 0 actual = r.get('actual_cost') or 0 r['variance'] = budget - actual columns = ['number', 'title', 'organization', 'department', 'budget', 'actual_cost', 'variance', 'status', 'starting_date'] elif report_name == 'user_activity_log': import json as _json2 log_conditions = ["v.ref_doctype IN ('SB Process', 'SB Order', 'SB Incident', 'SB Idea', 'SB Organization')"] log_values = {} if from_date and to_date: log_conditions.append('v.creation BETWEEN %(from_date)s AND %(to_date)s') log_values['from_date'] = from_date log_values['to_date'] = to_date + ' 23:59:59' elif from_date: log_conditions.append('v.creation >= %(from_date)s') log_values['from_date'] = from_date elif to_date: log_conditions.append('v.creation <= %(to_date)s') log_values['to_date'] = to_date + ' 23:59:59' raw = frappe.db.sql(""" SELECT v.creation AS timestamp, v.owner AS user, v.ref_doctype AS entity_type, v.docname AS entity_name, v.data FROM `tabVersion` v WHERE {conditions} ORDER BY v.creation DESC LIMIT 300 """.format(conditions=' AND '.join(log_conditions)), log_values, as_dict=True) rows = [] for row in raw: try: changed = _json2.loads(row.get('data') or '{}').get('changed', []) summary = ', '.join(f[0] for f in changed) if changed else 'Modified' except Exception: summary = 'Modified' rows.append({ 'timestamp': str(row.get('timestamp', '')), 'user': row.get('user', ''), 'entity_type': row.get('entity_type', ''), 'entity_name': row.get('entity_name', ''), 'changes': summary, }) columns = ['timestamp', 'user', 'entity_type', 'entity_name', 'changes'] else: frappe.throw("Unknown report: {}".format(report_name)) return {'columns': columns, 'rows': rows} @frappe.whitelist() def get_process_trend(organization=None, months=6): """ Return monthly Pass/Fail/Unknown counts for the last N months. Used by Level 2 Owner dashboard trend charts (Section 12.2). """ months = min(int(months), 24) organization = _user_organization() or organization org_filter = '' params = {'months': months} if organization: org_filter = 'AND p.organization = %(org)s' params['org'] = organization return frappe.db.sql(""" SELECT DATE_FORMAT(cl.cycle_date, '%%Y-%%m') AS month, cl.status, COUNT(*) AS count FROM `tabSB Process Cycle Log` cl JOIN `tabSB Process` p ON p.name = cl.parent WHERE cl.cycle_date >= DATE_SUB(CURDATE(), INTERVAL %(months)s MONTH) {org_filter} GROUP BY month, cl.status ORDER BY month ASC """.format(org_filter=org_filter), params, as_dict=True) @frappe.whitelist() def get_process_year_end(year=None, organization=None): """ Rule 11.1-6: Return full status history and total accumulated cost per process for a given year. year defaults to the current calendar year. """ from datetime import date as _date year = int(year or _date.today().year) filters_json = {} if organization: filters_json['organization'] = organization filters_json['year'] = year return get_report_data('process_year_end', filters_json) # ── Delegation ───────────────────────────────────────────────────────────────── _DELEGATABLE_FUNCTIONS = {'incident_initial_evaluation', 'idea_initial_evaluation'} @frappe.whitelist() def create_delegation(manager, functions_json, organization=None, remarks=None, incident_name=None, idea_name=None): """Users (Auditors / Managers / Admins) delegate Initial Evaluation to another Manager.""" import json as _json user = frappe.session.user roles = frappe.get_roles(user) category = frappe.db.get_value('User', user, 'custom_category') if not any(r in roles for r in ('SB Auditor', 'SB Admin', 'SB Manager', 'System Manager')) and category not in ('Auditor', 'Manager', 'Admin'): frappe.throw('You are not authorized to create delegations.') if user == manager: frappe.throw('You cannot delegate to yourself.') # Ensure recipient is a Manager mgr_roles = frappe.get_roles(manager) mgr_cat = frappe.db.get_value('User', manager, 'custom_category') if 'SB Manager' not in mgr_roles and mgr_cat != 'Manager': frappe.throw('Delegations can only be sent to Managers.') functions = _json.loads(functions_json) if isinstance(functions_json, str) else functions_json functions = [f for f in functions if f in _DELEGATABLE_FUNCTIONS] if not functions: frappe.throw('Select at least one valid function to delegate (Incident Initial Evaluation, Idea Initial Evaluation).') doc = frappe.get_doc({ 'doctype': 'SB Delegation', 'auditor': user, 'manager': manager, 'organization': organization or '', 'incident_initial_evaluation': 1 if 'incident_initial_evaluation' in functions else 0, 'idea_initial_evaluation': 1 if 'idea_initial_evaluation' in functions else 0, 'incident_name': incident_name or '', 'idea_name': idea_name or '', 'remarks': remarks or '', 'status': 'Pending', }) doc.append('delegation_log', { 'action': 'Created', 'actor': user, 'timestamp': frappe.utils.now_datetime(), 'remarks': remarks or '', }) doc.insert(ignore_permissions=True) frappe.db.commit() func_labels = [] if 'incident_initial_evaluation' in functions: label = 'Incident Initial Evaluation' if incident_name: label += f' (Incident: {incident_name})' func_labels.append(label) if 'idea_initial_evaluation' in functions: label = 'Idea Initial Evaluation' if idea_name: label += f' (Idea: {idea_name})' func_labels.append(label) _send_notification( manager, 'New delegation request from {}: {}'.format(user, ', '.join(func_labels)), 'SB Delegation', doc.name ) return doc.as_dict() @frappe.whitelist() def respond_to_delegation(name, response, remarks=None): """Manager accepts/rejects a pending delegation. BRD 4.3 — 'via their inbox'.""" if response not in ('Accepted', 'Rejected'): frappe.throw('Invalid response.') doc = frappe.get_doc('SB Delegation', name) user = frappe.session.user if doc.manager != user: frappe.throw('Only the delegated Manager may respond to this delegation.') if doc.status != 'Pending': frappe.throw('This delegation is no longer pending.') doc.status = response doc.responded_on = frappe.utils.now_datetime() doc.append('delegation_log', { 'action': response, 'actor': user, 'timestamp': frappe.utils.now_datetime(), 'remarks': remarks or '', }) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() _send_notification(doc.auditor, 'Delegation {} by {}'.format(response.lower(), user), 'SB Delegation', name) return doc.as_dict() @frappe.whitelist() def revoke_delegation(name, remarks=None): """Auditor or Delegator (or Admin) revokes at any time.""" doc = frappe.get_doc('SB Delegation', name) user = frappe.session.user is_admin = any(r in frappe.get_roles(user) for r in ('SB Admin', 'System Manager')) if doc.auditor != user and not is_admin: frappe.throw('Only the delegating user (or an Admin) may revoke this delegation.') if doc.status not in ('Pending', 'Accepted'): frappe.throw('Only Pending or Accepted delegations can be revoked.') doc.status = 'Revoked' doc.revoked_on = frappe.utils.now_datetime() doc.append('delegation_log', { 'action': 'Revoked', 'actor': user, 'timestamp': frappe.utils.now_datetime(), 'remarks': remarks or '', }) doc.flags.ignore_permissions = True doc.save(ignore_permissions=True) frappe.db.commit() _send_notification(doc.manager, 'Delegation revoked by {}'.format(user), 'SB Delegation', name) return doc.as_dict() @frappe.whitelist() def get_delegations(): """Role-scoped list of delegations for the Delegations page.""" user = frappe.session.user roles = frappe.get_roles(user) if any(r in roles for r in ('SB Admin', 'System Manager')): filters = {'custom_is_deleted': 0} elif 'SB Auditor' in roles or 'SB Manager' in roles: filters = { 'custom_is_deleted': 0, 'name': ['in', [d.name for d in frappe.get_all('SB Delegation', filters={'custom_is_deleted': 0}, or_filters={'auditor': user, 'manager': user}, ignore_permissions=True)]] } else: return [] return frappe.get_all('SB Delegation', filters=filters, fields=['name', 'number', 'auditor', 'manager', 'organization', 'incident_initial_evaluation', 'idea_initial_evaluation', 'status', 'requested_on', 'responded_on', 'revoked_on'], order_by='requested_on desc', ignore_permissions=True) @frappe.whitelist() def get_delegation(name): """Single-doc fetch (incl. delegation_log audit trail) with an ownership check.""" doc = frappe.get_doc('SB Delegation', name, ignore_permissions=True) user = frappe.session.user roles = frappe.get_roles(user) if doc.auditor != user and doc.manager != user and \ not any(r in roles for r in ('SB Admin', 'System Manager')): frappe.throw('Not permitted to view this delegation.') return doc.as_dict() @frappe.whitelist() def get_my_delegated_functions(): """Flat, org-agnostic list of function keys the calling Manager currently holds (any Accepted delegation, any organization) — used for frontend permission checks.""" rows = frappe.get_all('SB Delegation', filters={'manager': frappe.session.user, 'status': 'Accepted', 'custom_is_deleted': 0}, fields=['incident_initial_evaluation', 'idea_initial_evaluation'], ignore_permissions=True) funcs = set() for r in rows: if r.incident_initial_evaluation: funcs.add('incident_initial_evaluation') if r.idea_initial_evaluation: funcs.add('idea_initial_evaluation') return list(funcs) @frappe.whitelist() def get_my_delegated_organizations(): """Per-function organization scope for the calling Manager's Accepted delegations.""" rows = frappe.get_all('SB Delegation', filters={'manager': frappe.session.user, 'status': 'Accepted', 'custom_is_deleted': 0}, fields=['organization', 'incident_initial_evaluation', 'idea_initial_evaluation'], ignore_permissions=True) keys = ('incident_initial_evaluation', 'idea_initial_evaluation') orgs = {k: set() for k in keys} unrestricted = {k: False for k in keys} for r in rows: for k in keys: if r.get(k): if r.organization: orgs[k].add(r.organization) else: unrestricted[k] = True return {k: (None if unrestricted[k] else sorted(orgs[k])) for k in keys} # ── Inbox / Task Center ─────────────────────────────────────────────────────── @frappe.whitelist() def get_inbox_items(): """Return role-specific task items for the inbox page.""" from datetime import date as _dt_date, timedelta as _dt_timedelta from sidrah_business.tasks import _get_org_incident_windows, _TRIAGE_WINDOW_DEFAULT, _RESOLUTION_WINDOW_DEFAULT user = frappe.session.user roles = frappe.get_roles(user) items = [] _incident_windows = None # lazily loaded — only Auditor/Manager ever need it def _incident_resolution_label(organization, modified, fallback_label): """Rule 11.3-5 — flag as overdue once past the org's configured resolution window; otherwise fall back to the plain 'action required' label.""" nonlocal _incident_windows if _incident_windows is None: _incident_windows = _get_org_incident_windows() _, resolution_days = _incident_windows.get( organization, (_TRIAGE_WINDOW_DEFAULT, _RESOLUTION_WINDOW_DEFAULT)) if frappe.utils.getdate(modified) < _dt_date.today() - _dt_timedelta(days=resolution_days): return 'Resolution Overdue ({}+ days)'.format(resolution_days) return fallback_label is_auditor = any(r in roles for r in ['SB Auditor', 'SB Admin', 'System Manager']) is_owner = any(r in roles for r in ['SB Owner', 'SB Admin', 'System Manager']) is_manager = 'SB Manager' in roles is_employee = 'SB Employee' in roles if is_manager: for d in frappe.get_list('SB Delegation', filters={'manager': user, 'status': 'Pending'}, fields=['name', 'auditor', 'organization', 'incident_initial_evaluation', 'idea_initial_evaluation', 'requested_on'], order_by='requested_on desc', ignore_permissions=True): labels = [] if d.incident_initial_evaluation: labels.append('Incident Initial Evaluation') if d.idea_initial_evaluation: labels.append('Idea Initial Evaluation') items.append({ 'type': 'delegation', 'action': 'Delegation Pending Response', 'name': d.name, 'number': d.name, 'title': 'Delegated by {}: {}'.format(d.auditor, ', '.join(labels)), 'organization': d.organization or '', 'date': str(d.requested_on or ''), 'requires_action': True, }) # Delegated Incident Initial Evaluation for Manager inc_delegations = frappe.get_all('SB Delegation', filters={'manager': user, 'status': 'Accepted', 'incident_initial_evaluation': 1}, fields=['organization'], ignore_permissions=True) if inc_delegations: has_global_inc = any(not d.organization for d in inc_delegations) inc_filters = {'workflow_state': 'Initial Evaluation'} if not has_global_inc: inc_filters['organization'] = ['in', [d.organization for d in inc_delegations if d.organization]] delegated_incidents = frappe.get_list('SB Incident', filters=inc_filters, fields=['name', 'number', 'title', 'date', 'organization'], order_by='date asc', ignore_permissions=True) for r in delegated_incidents: items.append({ 'type': 'incident', 'action': 'Delegated Triage Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(r.date or ''), 'requires_action': True, }) # Delegated Idea Initial Evaluation for Manager idea_delegations = frappe.get_all('SB Delegation', filters={'manager': user, 'status': 'Accepted', 'idea_initial_evaluation': 1}, fields=['organization'], ignore_permissions=True) if idea_delegations: has_global_idea = any(not d.organization for d in idea_delegations) idea_filters = {'workflow_state': 'Initial Evaluation'} if not has_global_idea: idea_filters['organization'] = ['in', [d.organization for d in idea_delegations if d.organization]] delegated_ideas = frappe.get_list('SB Idea', filters=idea_filters, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True) for r in delegated_ideas: items.append({ 'type': 'idea', 'action': 'Delegated Evaluation Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # Incidents awaiting Auditor triage (Assign to Manager / Reject) if is_auditor: untriaged = frappe.get_list('SB Incident', filters={'workflow_state': 'Initial Evaluation'}, fields=['name', 'number', 'title', 'date', 'organization'], order_by='date asc', ignore_permissions=True) for r in untriaged: items.append({ 'type': 'incident', 'action': 'Triage Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(r.date or ''), 'requires_action': True, }) # Escalated processes → Auditor + Owner if is_auditor or is_owner: escalated_proc = frappe.get_list('SB Process', filters={'escalation_level': ['>=', 2]}, fields=['name', 'number', 'title', 'organization', 'status'], ignore_permissions=True) for r in escalated_proc: items.append({ 'type': 'process', 'action': 'Escalated — Action Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) escalated_ord = frappe.get_list('SB Order', filters={'escalation_level': ['>=', 2]}, fields=['name', 'number', 'title', 'organization', 'status'], ignore_permissions=True) for r in escalated_ord: items.append({ 'type': 'order', 'action': 'Escalated — Action Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # Auditor: ideas needing evaluation if is_auditor: # Ideas in Initial Evaluation ideas_eval = frappe.get_list('SB Idea', filters={'workflow_state': 'Initial Evaluation'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True) for r in ideas_eval: items.append({ 'type': 'idea', 'action': 'Evaluation Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # Incidents awaiting Resolution complete incidents_resolution = frappe.get_list('SB Incident', filters={'workflow_state': 'working on resolution'}, fields=['name', 'number', 'title', 'organization', 'date', 'modified'], ignore_permissions=True) for r in incidents_resolution: items.append({ 'type': 'incident', 'action': _incident_resolution_label(r.organization, r.modified, 'Resolution Required'), 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(r.date or ''), 'requires_action': True, }) # Assigned processes/orders → Employee or Manager (actor) if is_employee or is_manager: my_processes = frappe.get_list('SB Process', filters={'actor': user, 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization', 'cycle'], ignore_permissions=True) for r in my_processes: items.append({ 'type': 'process', 'action': 'Status Update Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # Orders in Pending Actor state — check both actor and assigned_to fields seen_orders = set() for fld in ('actor', 'assigned_to'): pending_orders = frappe.get_list('SB Order', filters={fld: user, 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization', 'stage'], ignore_permissions=True) for r in pending_orders: if r.name not in seen_orders: seen_orders.add(r.name) items.append({ 'type': 'order', 'action': 'Stage Update Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # Assigned incidents/ideas awaiting this specific Manager's solution/proposal # (Employee has no action here — Submit is their only Incident/Idea action) if is_manager: my_incidents = frappe.get_list('SB Incident', filters={'assigned_to': user, 'workflow_state': ['in', ['Accept & Under Progress', 'working on resolution']]}, fields=['name', 'number', 'title', 'organization', 'date', 'modified'], ignore_permissions=True) for r in my_incidents: items.append({ 'type': 'incident', 'action': _incident_resolution_label(r.organization, r.modified, 'Resolution Required'), 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(r.date or ''), 'requires_action': True, }) my_ideas = frappe.get_list('SB Idea', filters={'assigned_to': user, 'workflow_state': 'Accept & Under Progress'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True) for r in my_ideas: items.append({ 'type': 'idea', 'action': 'Proposal Required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', 'requires_action': True, }) # ── Approaching cycle deadline ──────────────────────────────────────────── # Actors whose cycle end_date is within the next 3 days and have not yet submitted if is_employee or is_manager: from datetime import date as _dt, timedelta as _td _today = _dt.today() upcoming_cycles = frappe.get_all('SB Cycle', filters={'status': 'Active', 'end_date': ['between', [str(_today), str(_today + _td(days=3))]]}, fields=['name', 'end_date'], ignore_permissions=True ) for cyc in upcoming_cycles: _deadline = frappe.utils.getdate(cyc.end_date) _days = (_deadline - _today).days _dlabel = 'today' if _days == 0 else 'in {} day{}'.format(_days, 's' if _days != 1 else '') _logged_proc = set(r[0] for r in frappe.db.sql( "SELECT DISTINCT parent FROM `tabSB Process Cycle Log` WHERE cycle=%s", cyc.name, as_list=True )) for r in frappe.get_all('SB Process', filters={'actor': user, 'cycle': cyc.name, 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): if r.name not in _logged_proc: items.append({ 'type': 'process', 'requires_action': True, 'action': 'Cycle deadline {} — submit status update'.format(_dlabel), 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(cyc.end_date), }) _logged_ord = set(r[0] for r in frappe.db.sql( "SELECT DISTINCT parent FROM `tabSB Order Cycle Log` WHERE cycle=%s", cyc.name, as_list=True )) _seen_deadline_ord = set() for _fld in ('actor', 'assigned_to'): for r in frappe.get_all('SB Order', filters={_fld: user, 'cycle': cyc.name, 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): if r.name not in _logged_ord and r.name not in _seen_deadline_ord: _seen_deadline_ord.add(r.name) items.append({ 'type': 'order', 'requires_action': True, 'action': 'Cycle deadline {} — submit stage update'.format(_dlabel), 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(cyc.end_date), }) # ── Unknown auto-assigned — actor acknowledgement (informational) ───────── # Only while still Pending Actor — once the process moves on (comment/review/study), # the Actor already saw it get triaged and a stale reminder no longer applies. if is_employee or is_manager: for r in frappe.get_all('SB Process', filters={'actor': user, 'status': 'Unknown', 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): items.append({ 'type': 'process', 'requires_action': False, 'action': 'Status auto-marked Unknown — cycle missed', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', }) # ── Unknown/Level-1 — auditor follow-up ────────────────────────────────── # Same rule — only while still Pending Actor; past that it's already in the # normal Auditor/Manager workflow queue (Comment Required, Review, etc.) if is_auditor: for r in frappe.get_all('SB Process', filters={'status': 'Unknown', 'escalation_level': 1, 'workflow_state': 'Pending Actor'}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): items.append({ 'type': 'process', 'requires_action': True, 'action': 'Unknown status — Level 1 follow-up needed', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', }) # ── Auditor comment on process/order — notify actor ─────────────────────── if is_employee or is_manager: for r in frappe.get_all('SB Process', filters={'actor': user, 'workflow_state': 'Pending Actor', 'auditor_comment': ['!=', '']}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): items.append({ 'type': 'process', 'requires_action': False, 'action': 'Auditor left a comment — review before submitting', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', }) _seen_comm_ord = set() for _fld in ('actor', 'assigned_to'): for r in frappe.get_all('SB Order', filters={_fld: user, 'workflow_state': 'Pending Actor', 'auditor_comment': ['!=', '']}, fields=['name', 'number', 'title', 'organization'], ignore_permissions=True ): if r.name not in _seen_comm_ord: _seen_comm_ord.add(r.name) items.append({ 'type': 'order', 'requires_action': False, 'action': 'Auditor left a comment — review before submitting', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': '', }) # ── Auditor/Owner comment on incident — notify assigned manager ─────────── if is_manager: for r in frappe.get_all('SB Incident', filters={ 'assigned_to': user, 'auditor_comment': ['!=', ''], 'workflow_state': ['in', ['Accept & Under Progress', 'working on resolution']], }, fields=['name', 'number', 'title', 'organization', 'date'], ignore_permissions=True ): items.append({ 'type': 'incident', 'requires_action': False, 'action': 'Auditor/Owner comment — review required', 'name': r.name, 'number': r.number or r.name, 'title': r.title, 'organization': r.organization, 'date': str(r.date or ''), }) # A user restricted to one Organization only sees inbox items for it — applies # to any role, not just Employees; unrestricted users (no custom_organization # set, e.g. Admins) see everything, same as every other list in the app. my_org = _user_organization() if my_org: items = [i for i in items if not i.get('organization') or i['organization'] == my_org] return items # ── Comments & Activity (permission-safe wrappers) ──────────────────────────── @frappe.whitelist() def get_doc_comments(doctype, docname): """Return all Comment records for a doc, accessible to any authenticated user.""" return frappe.get_all( 'Comment', filters={ 'reference_doctype': doctype, 'reference_name': docname, 'comment_type': 'Comment', }, fields=['name', 'owner', 'creation', 'content', 'comment_by', 'comment_email'], order_by='creation asc', limit=200, ignore_permissions=True, ) @frappe.whitelist() def post_doc_comment(doctype, docname, content): """Create a Comment record for any authenticated user.""" if not content or not content.strip(): frappe.throw('Comment cannot be empty') comment = frappe.get_doc({ 'doctype': 'Comment', 'comment_type': 'Comment', 'reference_doctype': doctype, 'reference_name': docname, 'content': content.strip(), 'comment_by': frappe.session.user, 'comment_email': frappe.session.user, }) comment.flags.ignore_permissions = True comment.insert(ignore_permissions=True) frappe.db.commit() return comment.as_dict() @frappe.whitelist() def get_doc_versions(doctype, docname): """Return Version (change history) records, bypassing System Manager restriction.""" return frappe.get_all( 'Version', filters={'ref_doctype': doctype, 'docname': docname}, fields=['name', 'owner', 'creation', 'data'], order_by='creation desc', limit=50, ignore_permissions=True, ) # ── Notifications ───────────────────────────────────────────────────────────── @frappe.whitelist() def _filter_notifications_by_org(rows): """A user restricted to one Organization shouldn't see a notification for a record that belongs to a different one — even though the notification itself was validly addressed to them (e.g. they were once assigned there), opening it would hit the same cross-org permission wall as acting on it directly, so hide it instead.""" my_org = _user_organization() if not my_org: return rows out = [] for r in rows: org_field = _ORG_SCOPED_FIELD.get(r.get('document_type')) if not org_field: out.append(r) continue doc_org = frappe.db.get_value(r['document_type'], r['document_name'], org_field) if not doc_org or doc_org == my_org: out.append(r) return out @frappe.whitelist() def get_notification_count(): try: result = frappe.get_all( 'Notification Log', filters={'for_user': frappe.session.user, 'read': 0}, fields=['name', 'document_type', 'document_name'], ignore_permissions=True ) return len(_filter_notifications_by_org(result)) except Exception: return 0 @frappe.whitelist() def get_notifications(limit=20): try: rows = frappe.get_all( 'Notification Log', filters={'for_user': frappe.session.user}, fields=['name', 'subject', 'type', 'document_type', 'document_name', 'read', 'creation'], order_by='creation desc', limit=int(limit) * 2, ignore_permissions=True ) return _filter_notifications_by_org(rows)[:int(limit)] except Exception: return [] @frappe.whitelist() def mark_notifications_read(): try: frappe.db.sql( "UPDATE `tabNotification Log` SET `read` = 1 WHERE `for_user` = %(user)s AND `read` = 0", {'user': frappe.session.user} ) frappe.db.commit() except Exception: pass return 'ok' @frappe.whitelist() def mark_notification_read(name): """Mark a single Notification Log row read — used when the user clicks through to the entity from the notification bell, so it stops counting as unread.""" try: if frappe.db.get_value('Notification Log', name, 'for_user') == frappe.session.user: frappe.db.set_value('Notification Log', name, 'read', 1, update_modified=False) frappe.db.commit() except Exception: pass return 'ok' @frappe.whitelist() def update_process_doc_status(name, status): """Allow any authenticated user to update the status field of SB Process.""" if status and not frappe.db.exists('Process Statuses', status): frappe.throw(_('Invalid status: {}').format(status)) frappe.db.set_value('SB Process', name, 'status', status or None) frappe.db.commit() return {'status': status} # ── Internal helpers ────────────────────────────────────────────────────────── def _send_notification(user, subject, doctype, docname): """In-app bell notification only. Real emails for each workflow step are owned by the matching Frappe Notification (Email Alert) record — see _fire_state_alert() below — so a step's email content/recipients can be edited in Frappe Desk without touching code, and nothing here double-sends the same email.""" try: frappe.get_doc({ 'doctype': 'Notification Log', 'subject': subject, 'for_user': user, 'type': 'Alert', 'document_type': doctype, 'document_name': docname, }).insert(ignore_permissions=True) frappe.db.commit() except Exception: pass def _fire_state_alert(doctype, doc): """Send the Frappe Notification (Email Alert) record for whatever state `doc` is in right now, named ' - ' (see the rebuild_workflow_notifications patch for the full set). Each is a real event='Value Change' / value_changed='workflow_state' record with a `doc.workflow_state == "..."` condition — editable in Desk like any other Notification — but Frappe's own automatic dispatch can't be the thing that fires it: workflow_state is advanced via frappe.db.set_value specifically to dodge Frappe's Workflow permission check, so it never trips a native trigger. Our workflow functions call this explicitly, right after workflow_state is set, which is the one place that reliably knows the transition just happened; the record's own `condition` is still honored here, so editing it in Desk changes real behavior.""" state = getattr(doc, 'workflow_state', None) if not state: return alert_name = '{} - {}'.format(doctype, state) if not frappe.db.exists('Notification', alert_name): return try: alert = frappe.get_cached_doc('Notification', alert_name) if not alert.enabled: return if alert.condition and not frappe.safe_eval(alert.condition, None, {'doc': doc.as_dict()}): return alert.send(doc) except Exception: frappe.log_error(title='Sidrah alert dispatch failed: {}'.format(alert_name), message=frappe.get_traceback()) def _notify_escalation(doctype, docname, organization, entity_label): owners = frappe.get_all('Has Role', filters={'role': ['in', ['SB Owner', 'SB Admin', 'System Manager']]}, fields=['parent'], ignore_permissions=True) for owner in owners: _send_notification( owner.parent, '{} Escalated — Immediate Attention Required'.format(entity_label), doctype, docname ) def _notify_org_role(role, organization, subject, doctype, docname, exclude=None): """Notify every holder of `role` who can see this organization (pinned to it, or unrestricted) — used wherever a workflow step hands off to a role rather than one specific person, since there's no single named recipient to look up on the doc. Never notifies the user who triggered the action (no self-notification).""" holders = frappe.get_all('Has Role', filters={'role': role}, fields=['parent'], ignore_permissions=True) seen = set() for h in holders: recipient = h.parent if recipient in seen or recipient == exclude or recipient == frappe.session.user: continue seen.add(recipient) their_org = frappe.db.get_value('User', recipient, 'custom_organization') if their_org and their_org != organization: continue _send_notification(recipient, subject, doctype, docname) def _notify_org_managers(organization, subject, doctype, docname, exclude=None): """Notify every SB Manager who can see this organization — used where BRD 13.1 names 'Manager' as a recipient alongside a specific Actor, since a single Actor field doesn't capture organizational oversight.""" _notify_org_role('SB Manager', organization, subject, doctype, docname, exclude) def _notify_org_auditors(organization, subject, doctype, docname, exclude=None): """Notify every SB Auditor who can see this organization — the default recipient whenever a workflow step hands a decision to 'whoever is auditing this org', with no single assigned individual to target.""" _notify_org_role('SB Auditor', organization, subject, doctype, docname, exclude) # Each transactional doctype's `status` field Links to its own lookup master # (e.g. SB Incident.status -> Incident Statuses), and that master carries a # `notify_manager` checkbox + its own Active/Inactive `status`. This is the # single place that checks both before paging the record's Manager/Actor. _STATUS_MASTER_DOCTYPE = { 'SB Incident': 'Incident Statuses', 'SB Idea': 'Idea Statuses', 'SB Order': 'Order Statuses', 'SB Process': 'Process Statuses', } def _notify_manager_on_status(doctype, doc): """Notify the record's Manager/Actor on a status change, but only when the new status's master record has 'Notify Manager' checked AND that master record is itself Active — an Inactive or non-notifying status stays silent.""" master_doctype = _STATUS_MASTER_DOCTYPE.get(doctype) status_value = getattr(doc, 'status', None) if not master_doctype or not status_value: return master = frappe.db.get_value( master_doctype, status_value, ['notify_manager', 'status'], as_dict=True ) if not master or not master.notify_manager or master.status != 'Active': return recipient = getattr(doc, 'assigned_to', None) or getattr(doc, 'actor', None) if not recipient: return _send_notification( recipient, '{} status updated to {}: {}'.format(doctype.replace('SB ', ''), status_value, doc.title or doc.name), doctype, doc.name ) def _log_version(doctype, name, changed): """Workflow transitions move `workflow_state` via frappe.db.set_value (to bypass doc.save()'s Workflow permission check — see apply_*_workflow) instead of doc.save(), so track_changes=1 never picks them up and they're invisible in the Activity Log even though they're the main story of the record's history. Record a Version entry by hand to match what doc.save() would have logged.""" import json as _json try: frappe.get_doc({ 'doctype': 'Version', 'ref_doctype': doctype, 'docname': name, 'data': _json.dumps({'changed': changed, 'added': [], 'removed': [], 'row_changed': []}), }).insert(ignore_permissions=True) except Exception: pass def _user_group_company(): """The Group Company this user is restricted to, or None when unrestricted (Administrator / global CEO).""" from sidrah_business.soft_delete import is_ceo_user, get_user_group_company user = frappe.session.user if user == 'Administrator': return None co = get_user_group_company(user) if not co and is_ceo_user(user): return None return co def _user_organization(): """The Organization this user is restricted to, or None when unrestricted (Administrator / global CEO).""" from sidrah_business.soft_delete import is_ceo_user user = frappe.session.user if user == 'Administrator': return None org = frappe.db.get_value('User', user, 'custom_organization') or None if not org and is_ceo_user(user): return None return org # Field that carries the owning Organization, per org-scoped doctype _ORG_SCOPED_FIELD = { 'SB Incident': 'organization', 'SB Idea': 'organization', 'SB Order': 'organization', 'SB Process': 'organization', 'SB Cycle': 'organization', 'SB KPI Input': 'organization', 'Managements': 'organization', 'Branch': 'organization', 'Department': 'custom_organization', 'SB Delegation': 'organization', } def _organization_filter(): """Filter dict to merge into frappe.get_list/get_all filters — restricts results to the calling user's Group Company and Organization when they're scoped.""" filters = {} my_co = _user_group_company() if my_co: filters['group_company'] = ['in', [my_co, '', None]] my_org = _user_organization() if my_org: filters['organization'] = ['in', [my_org, '', None]] return filters def _apply_organization_scope(doctype, data): """A user restricted to a Group Company or Organization can only create/save records under it.""" my_co = _user_group_company() if my_co: data['group_company'] = my_co elif not data.get('group_company'): org = data.get('organization') or data.get('custom_organization') if org and frappe.db.exists('SB Organization', org): org_co = frappe.db.get_value('SB Organization', org, 'group_company') if org_co: data['group_company'] = org_co field = _ORG_SCOPED_FIELD.get(doctype) my_org = _user_organization() if field and my_org: data[field] = my_org return data def _enforce_same_organization(doctype, record_org, record_co=None): """Enforces Group Company and Organization isolation on document read/action.""" from sidrah_business.soft_delete import is_platform_admin, is_group_company_admin user = frappe.session.user if user == 'Administrator': return my_co = _user_group_company() if my_co and record_co and record_co != my_co: frappe.throw( 'You do not have permission to access this {} — it belongs to a different Group Company.'.format(doctype), frappe.PermissionError ) my_org = _user_organization() if my_org and record_org and record_org != my_org: frappe.throw( 'You do not have permission to access this {} — it belongs to a different organization.'.format(doctype), frappe.PermissionError ) def _has_active_delegation(user, function_key, organization=None): """True if `user` holds an Accepted SB Delegation granting `function_key`, either global (blank organization) or matching `organization`.""" rows = frappe.get_all('SB Delegation', filters={'manager': user, 'status': 'Accepted', function_key: 1}, fields=['organization'], ignore_permissions=True) return any((not r.organization) or r.organization == organization for r in rows) def _enforce_process_write_permission(organization): """Auditors and Admins may create/edit SB Process records. Delegation is no longer available for Processes.""" roles = frappe.get_roles(frappe.session.user) if any(r in roles for r in ('SB Admin', 'SB Auditor', 'SB Owner', 'System Manager')): return frappe.throw('You do not have permission to create or edit Processes.') def _enforce_incident_idea_write_permission(doctype, name): """Employee may draft/edit their own report only while it is still 'New'; once it has been Submitted, further changes must go through the workflow actions (Auditor/Manager).""" roles = frappe.get_roles(frappe.session.user) if any(r in roles for r in ('SB Admin', 'SB Auditor', 'SB Manager', 'System Manager')): return if 'SB Employee' not in roles: frappe.throw('You do not have permission to edit this {}.'.format(doctype)) if name: current_state = frappe.db.get_value(doctype, name, 'workflow_state') if current_state and current_state != 'New': frappe.throw('This {} has already been submitted and can no longer be edited directly.'.format(doctype)) def _enforce_order_write_permission(): """Orders are Auditor/Manager-owned documents — Employees only ever touch them via the 'Update Status' workflow action (as the assigned Actor), never via direct field edits.""" roles = frappe.get_roles(frappe.session.user) if not any(r in roles for r in ('SB Admin', 'SB Auditor', 'SB Manager', 'System Manager')): frappe.throw('You do not have permission to edit Orders.') _PROCESS_AUDITOR_OR_MANAGER_ACTIONS = { 'Submit for Review', 'Add Comment', 'Move to Study', 'Start New Cycle', 'Close Process', } def _enforce_process_action_permission(action): """Every Process workflow action except 'Update Status' is Auditor-or-Manager.""" roles = frappe.get_roles(frappe.session.user) if any(r in roles for r in ('SB Admin', 'System Manager')): return if action in _PROCESS_AUDITOR_OR_MANAGER_ACTIONS: if not any(r in roles for r in ('SB Auditor', 'SB Manager')): frappe.throw("Only an Auditor or Manager can perform '{}'.".format(action), frappe.PermissionError) def _is_auditor_or_admin(user=None): """Check if user has Auditor or Admin privileges.""" user = user or frappe.session.user if user == 'Administrator': return True roles = frappe.get_roles(user) return any(r in roles for r in ('SB Auditor', 'SB Admin', 'System Manager')) _INCIDENT_IDEA_AUDITOR_ONLY_ACTIONS = {'Assign to Manager', 'Reject', 'Move to Outsourcing'} _INCIDENT_IDEA_MANAGER_ACTIONS = {'Submit Solution', 'Submit Proposal', 'Resolution complete'} def _enforce_incident_idea_action_permission(action, doctype=None, doc=None): """Submit is open to whoever created the report (incl. Employee); triage/assignment and the initial evaluation are Auditor-only unless a Manager holds an active delegation; the solution/proposal step needs Manager or Auditor.""" roles = frappe.get_roles(frappe.session.user) user = frappe.session.user if any(r in roles for r in ('SB Admin', 'System Manager')): return if action in _INCIDENT_IDEA_AUDITOR_ONLY_ACTIONS: if action in ('Assign to Manager', 'Reject'): if 'SB Auditor' in roles: return if 'SB Manager' in roles and doctype and doc: fn_key = 'incident_initial_evaluation' if doctype == 'SB Incident' else 'idea_initial_evaluation' if _has_active_delegation(user, fn_key, getattr(doc, 'organization', None)): return frappe.throw("Only an Auditor or a delegated Manager can perform '{}'.".format(action), frappe.PermissionError) elif action == 'Move to Outsourcing': if 'SB Auditor' not in roles: frappe.throw("Only an Auditor can perform '{}'.".format(action), frappe.PermissionError) elif action in _INCIDENT_IDEA_MANAGER_ACTIONS: if not any(r in roles for r in ('SB Manager', 'SB Auditor')): frappe.throw("Only a Manager or Auditor can perform '{}'.".format(action), frappe.PermissionError) def _enforce_order_action_permission(doc, action): """Submit for Review / Review-stage actions are Auditor/Manager (Owner may also act on Review-stage docs); Update Status is restricted to the assigned Actor unless privileged.""" roles = frappe.get_roles(frappe.session.user) if any(r in roles for r in ('SB Admin', 'System Manager')): return if action == 'Update Status': is_actor = frappe.session.user in (doc.actor, doc.assigned_to) if not is_actor and not any(r in roles for r in ('SB Auditor', 'SB Manager')): frappe.throw('You are not the assigned Actor for this Order.', frappe.PermissionError) elif action == 'Submit for Review': if not any(r in roles for r in ('SB Auditor', 'SB Manager')): frappe.throw('Only an Auditor or Manager can submit an Order for review.', frappe.PermissionError) else: # Move to Study / Start New Cycle / Close Order if not any(r in roles for r in ('SB Auditor', 'SB Manager', 'SB Owner')): frappe.throw("You do not have permission to perform '{}'.".format(action), frappe.PermissionError) # ── Generic permission-bypassing create / update for SB doctypes ───────────── _SB_ALLOWED_DOCTYPES = { 'SB Incident', 'SB Idea', 'SB Order', 'SB Process', 'SB Organization', 'SB Cycle', 'SB KPI Input', 'Managements', 'Department', 'Branch', 'Incident Types', 'Idea Types', 'Order Types', 'Process Types', 'Incident Class', 'Incident Field', 'Incident Statuses', 'Idea Statuses', 'Order Statuses', 'Process Statuses', 'SB Group Company', 'User', } # Identifying field(s) that must not change once a master-data record exists — the # frontend disables these on edit, this is the server-side backstop. _LOCKED_FIELDS_ON_UPDATE = { 'SB Organization': ('organization_name', 'acronym'), 'SB Cycle': ('name1',), 'Managements': ('name1',), 'Department': ('department_name',), 'Branch': ('branch',), 'Incident Types': ('name1',), 'Idea Types': ('name1',), 'Order Types': ('name1',), 'Process Types': ('name1',), 'Incident Statuses': ('name1',), 'Idea Statuses': ('name1',), 'Order Statuses': ('name1',), 'Process Statuses': ('name1',), 'Incident Class': ('name1',), 'Incident Field': ('name1',), # auditor_comment is a workflow-step action ("Add Comment"), not a freeform field — # locking it out of the generic update path forces it through apply_process_workflow, # which is the only place that's state-gated (Pending Comment) and gets logged as the # actual workflow event, instead of a silent edit that skips the step entirely. 'SB Process': ('auditor_comment',), } def _handle_create_user(data): from sidrah_business.soft_delete import is_platform_admin, is_admin_user, is_business_owner_user current_user = frappe.session.user is_plat_admin = is_platform_admin(current_user) is_adm = is_admin_user(current_user) is_biz = is_business_owner_user(current_user) if not (is_plat_admin or is_adm or is_biz or current_user == 'Administrator'): frappe.throw(_('Permission denied to create user.'), frappe.PermissionError) target_category = data.get('custom_category') # Admin cannot create CEO if is_adm and not is_plat_admin and not is_biz and current_user != 'Administrator': if target_category in ('CEO', 'Chairman'): frappe.throw(_('Admin is not allowed to create a CEO user.'), frappe.PermissionError) # Business Owner can create CEO and Auditor if is_biz and not is_plat_admin and current_user != 'Administrator': if target_category not in ('CEO', 'Auditor', 'X-Auditor'): frappe.throw(_('Business Owner is only allowed to create CEO and Auditor users.'), frappe.PermissionError) new_password = data.pop('new_password', None) email = data.get('email') if not email: frappe.throw(_('Email is required.')) data.pop('name', None) data['send_welcome_email'] = 0 data['user_type'] = 'System User' my_co = _user_group_company() if my_co and not data.get('custom_group_company'): data['custom_group_company'] = my_co user_doc = frappe.new_doc('User') user_doc.update(data) user_doc.insert(ignore_permissions=True) if new_password: from frappe.utils.password import update_password update_password(user_doc.name, new_password) from sidrah_business.setup_multitenancy import sync_user_group_company_permission sync_user_group_company_permission(user_doc) frappe.db.commit() return user_doc.as_dict() def _handle_update_user(name, data): from sidrah_business.soft_delete import is_platform_admin, is_admin_user, is_business_owner_user current_user = frappe.session.user is_plat_admin = is_platform_admin(current_user) is_adm = is_admin_user(current_user) is_biz = is_business_owner_user(current_user) if not (is_plat_admin or is_adm or is_biz or current_user == name or current_user == 'Administrator'): frappe.throw(_('Permission denied to update user.'), frappe.PermissionError) target_category = data.get('custom_category') if target_category and is_adm and not is_plat_admin and not is_biz and current_user != 'Administrator': if target_category in ('CEO', 'Chairman'): frappe.throw(_('Admin is not allowed to assign CEO role/category.'), frappe.PermissionError) if is_biz and not is_plat_admin and current_user != name and current_user != 'Administrator': if target_category and target_category not in ('CEO', 'Auditor', 'X-Auditor'): frappe.throw(_('Business Owner is only allowed to assign CEO or Auditor category.'), frappe.PermissionError) new_password = data.pop('new_password', None) data.pop('name', None) data.pop('email', None) user_doc = frappe.get_doc('User', name) user_doc.update(data) user_doc.flags.ignore_permissions = True user_doc.save(ignore_permissions=True) if new_password: from frappe.utils.password import update_password update_password(user_doc.name, new_password) from sidrah_business.setup_multitenancy import sync_user_group_company_permission sync_user_group_company_permission(user_doc) frappe.db.commit() return user_doc.as_dict() def validate_user_creation_permissions(doc, method=None): user = frappe.session.user if user in ('Administrator', 'Guest'): return from sidrah_business.soft_delete import is_platform_admin, is_admin_user, is_business_owner_user target_category = getattr(doc, 'custom_category', None) creator_roles = frappe.get_roles(user) is_plat_admin = is_platform_admin(user) is_biz = is_business_owner_user(user) is_adm = ('SB Admin' in creator_roles or 'System Manager' in creator_roles) and not is_plat_admin and not is_biz if is_adm: # Admin must not create a CEO user if target_category in ('CEO', 'Chairman'): frappe.throw(_('Admin is not allowed to create a CEO user.'), frappe.PermissionError) if is_biz and not is_plat_admin: # Business Owner can create CEO and Auditor if doc.is_new() and target_category not in ('CEO', 'Auditor', 'X-Auditor'): frappe.throw(_('Business Owner is only allowed to create CEO and Auditor users.'), frappe.PermissionError) @frappe.whitelist() def create_sb_doc(doctype, data_json): import json as _json if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) data = _json.loads(data_json) if isinstance(data_json, str) else data_json if doctype == 'User': return _handle_create_user(data) from sidrah_business.soft_delete import is_business_owner_user if is_business_owner_user() and doctype in ('SB Incident', 'SB Idea', 'SB Order', 'SB Process'): frappe.throw(_('Business Owners are not allowed to create {}.').format(doctype), frappe.PermissionError) data = _apply_organization_scope(doctype, data) if doctype == 'SB Process': _enforce_process_write_permission(data.get('organization')) elif doctype in ('SB Incident', 'SB Idea'): _enforce_incident_idea_write_permission(doctype, None) if not _is_auditor_or_admin(): data.pop('assigned_to', None) elif doctype == 'SB Order': _enforce_order_write_permission() if not _is_auditor_or_admin(): data.pop('assigned_to', None) data.pop('actor', None) data.pop('name', None) doc = frappe.get_doc({'doctype': doctype, **data}) doc.insert(ignore_permissions=True) frappe.db.commit() return doc.as_dict() @frappe.whitelist() def update_sb_doc(doctype, name, data_json): import json as _json if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) data = _json.loads(data_json) if isinstance(data_json, str) else data_json if doctype == 'User': return _handle_update_user(name, data) from sidrah_business.soft_delete import is_business_owner_user if is_business_owner_user() and doctype in ('SB Incident', 'SB Idea', 'SB Order', 'SB Process'): frappe.throw(_('Business Owners cannot modify main DocTypes. Only comments can be added.'), frappe.PermissionError) data = _apply_organization_scope(doctype, data) org_field = _ORG_SCOPED_FIELD.get(doctype) if org_field: _enforce_same_organization(doctype, frappe.db.get_value(doctype, name, org_field)) elif doctype == 'SB Organization': _enforce_same_organization(doctype, name) if doctype == 'SB Process': existing_org = frappe.db.get_value('SB Process', name, 'organization') _enforce_process_write_permission(data.get('organization', existing_org)) elif doctype in ('SB Incident', 'SB Idea'): _enforce_incident_idea_write_permission(doctype, name) elif doctype == 'SB Order': _enforce_order_write_permission() elif doctype == 'SB Delegation': st = frappe.db.get_value('SB Delegation', name, 'status') if st in ('Revoked', 'Rejected', 'Accepted'): frappe.throw('Delegation {} is {} and cannot be updated.'.format(name, st)) if doctype in ('SB Incident', 'SB Idea', 'SB Order', 'SB Process') and not _is_auditor_or_admin(): data.pop('assigned_to', None) data.pop('actor', None) data.pop('name', None) for f in _LOCKED_FIELDS_ON_UPDATE.get(doctype, ()): data.pop(f, None) doc = frappe.get_doc(doctype, name) comment_changed = ( doctype in ('SB Incident', 'SB Idea') and data.get('auditor_comment') and data.get('auditor_comment') != doc.auditor_comment ) for k, v in data.items(): if hasattr(doc, k): doc.set(k, v) doc.flags.ignore_permissions = True doc.flags.ignore_workflow = True doc.flags.ignore_links = True doc.save(ignore_permissions=True) frappe.db.commit() if doctype in _STATUS_MASTER_DOCTYPE and 'status' in data: _notify_manager_on_status(doctype, doc) # Owner/Auditor comment flows to the assigned Manager for acknowledgement (BRD 9.3/13.1) if comment_changed and getattr(doc, 'assigned_to', None): _send_notification( doc.assigned_to, 'New comment on {}: {}'.format(doctype.replace('SB ', ''), doc.title or name), doctype, name ) return doc.as_dict() @frappe.whitelist() def delete_sb_doc(doctype, name, reason=None, permanent=False): if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) from sidrah_business.soft_delete import soft_delete, is_admin_user, is_business_owner_user if is_business_owner_user() and doctype in ('SB Incident', 'SB Idea', 'SB Order', 'SB Process'): frappe.throw(_('Business Owners cannot delete main DocTypes.'), frappe.PermissionError) if permanent and (permanent == '1' or permanent is True or str(permanent).lower() == 'true'): if not is_admin_user(): frappe.throw(_("Only Administrators can permanently delete records."), frappe.PermissionError) frappe.flags.allow_permanent_delete = True try: frappe.delete_doc(doctype, name, ignore_permissions=True) frappe.db.commit() finally: frappe.flags.allow_permanent_delete = False return {'deleted': name, 'permanent': True} return soft_delete(doctype, name, reason=reason) @frappe.whitelist() def restore_sb_doc(doctype, name): if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) from sidrah_business.soft_delete import restore_doc return restore_doc(doctype, name) @frappe.whitelist() def disable_user(name): frappe.db.set_value('User', name, 'enabled', 0, update_modified=False) frappe.db.commit() return {'disabled': name} @frappe.whitelist() def get_sb_list(doctype, limit=500): if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) from sidrah_business.soft_delete import ( is_platform_admin, is_group_company_admin, is_manager_user, is_auditor_user, get_auditors_under_manager, is_ceo_user ) user = frappe.session.user my_org = _user_organization() my_co = _user_group_company() # 1. Root Administrator or unrestricted CEO: sees all records across system if user == 'Administrator' or (is_ceo_user(user) and not my_org and not my_co): return frappe.get_all(doctype, fields=['*'], limit_page_length=int(limit), ignore_permissions=True) # 2. Multi-tenant Group Company requirement if not my_co and not my_org and not is_platform_admin(user): return [] # 3. Query scoped to group company and organization if applicable filters = {} if my_co: filters['group_company'] = my_co if my_org: if doctype == 'SB Organization': filters['name'] = my_org elif doctype == 'Department': filters['custom_organization'] = my_org elif doctype in ('Managements', 'Branch', 'SB Cycle', 'SB Incident', 'SB Idea', 'SB Order', 'SB Process', 'SB KPI Input'): filters['organization'] = my_org rows = frappe.get_all(doctype, filters=filters, fields=['*'], limit_page_length=int(limit), ignore_permissions=True) # 4. Enforce strict org filter in memory for any doctypes that might have missing or extra rows if my_org: if doctype == 'SB Organization': rows = [r for r in rows if r.get('name') == my_org] elif doctype == 'Department': rows = [r for r in rows if r.get('custom_organization') == my_org] elif doctype in ('Managements', 'Branch', 'SB Cycle', 'SB Incident', 'SB Idea', 'SB Order', 'SB Process', 'SB KPI Input'): rows = [r for r in rows if r.get('organization') == my_org] # 5. Soft delete filtering for non-admins if not is_group_company_admin(user, my_co) and not is_platform_admin(user): rows = [r for r in rows if not r.get('custom_is_deleted')] # 6. Master doctypes: visible to users in this company and scoped org if doctype in ('SB Organization', 'Managements', 'Department', 'Branch', 'SB Cycle'): return rows # 7. Group Company Admin / Platform Admin: sees ALL data in their company and scoped org if is_group_company_admin(user, my_co) or is_platform_admin(user): return rows # 8. Manager: sees own data + data of all auditors and employees present under him (recursively) # Manager 1 sees only data present under him, NOT Manager 2's data! if is_manager_user(user): subs = get_subordinates_recursive(user) allowed_users = set([user] + subs) filtered = [] for r in rows: owner = r.get('owner') assigned_to = r.get('assigned_to') actor = r.get('actor') if owner in allowed_users or assigned_to in allowed_users or actor in allowed_users: filtered.append(r) return filtered # 9. Auditor: sees own data + data of all employees present under him! # Auditor 1 sees E1, E2, E3, E4. Auditor 1 CANNOT see Auditor 2's data or Auditor 2's employees! if is_auditor_user(user): subs = get_subordinates_recursive(user) allowed_users = set([user] + subs) filtered = [] for r in rows: owner = r.get('owner') assigned_to = r.get('assigned_to') actor = r.get('actor') if owner in allowed_users or assigned_to in allowed_users or actor in allowed_users: filtered.append(r) return filtered # 10. Employee: employee1 sees ONLY his own data (not to see data created by employee2)! filtered = [] for r in rows: owner = r.get('owner') assigned_to = r.get('assigned_to') actor = r.get('actor') if owner == user or assigned_to == user or actor == user: filtered.append(r) return filtered @frappe.whitelist() def get_sb_doc(doctype, name): if doctype not in _SB_ALLOWED_DOCTYPES: frappe.throw('Doctype not allowed: {}'.format(doctype)) doc = frappe.get_doc(doctype, name, ignore_permissions=True) from sidrah_business.soft_delete import ( is_platform_admin, is_group_company_admin, is_manager_user, is_auditor_user, get_subordinates_recursive ) user = frappe.session.user # CEO / Platform Admin can read everything across system if is_platform_admin(user): return doc.as_dict() my_co = _user_group_company() if my_co and doc.get('group_company') and doc.get('group_company') != my_co: frappe.throw('You do not have permission to access this {} — it belongs to a different Group Company.'.format(doctype), frappe.PermissionError) if not is_group_company_admin(user, my_co) and getattr(doc, 'custom_is_deleted', 0): frappe.throw(_("Document not found or access denied."), frappe.DoesNotExistError) # Master doctypes allowed within company if doctype in ('SB Organization', 'Managements', 'Department', 'Branch', 'SB Cycle'): return doc.as_dict() # Group Company Admin can see all in that company if is_group_company_admin(user, my_co): return doc.as_dict() # Manager can see if owner, assigned_to, actor is self or any subordinate (auditor or employee under him) if is_manager_user(user): subs = get_subordinates_recursive(user) allowed = set([user] + subs) if doc.get('owner') in allowed or doc.get('assigned_to') in allowed or doc.get('actor') in allowed: return doc.as_dict() frappe.throw('You do not have permission to access this record.', frappe.PermissionError) # Auditor can see if owner, assigned_to, actor is self or any employee under him if is_auditor_user(user): subs = get_subordinates_recursive(user) allowed = set([user] + subs) if doc.get('owner') in allowed or doc.get('assigned_to') in allowed or doc.get('actor') in allowed: return doc.as_dict() frappe.throw('You do not have permission to access this record.', frappe.PermissionError) # Employee / Regular user: only sees own data or assigned data if doc.get('owner') == user or doc.get('assigned_to') == user or doc.get('actor') == user: return doc.as_dict() frappe.throw('You do not have permission to access this record.', frappe.PermissionError) @frappe.whitelist() def get_group_companies(): """Returns list of SB Group Companies visible to current user.""" from sidrah_business.soft_delete import is_platform_admin if is_platform_admin(): return frappe.get_all("SB Group Company", filters={"custom_is_deleted": 0}, fields=["*"], ignore_permissions=True) my_co = _user_group_company() if my_co: return frappe.get_all("SB Group Company", filters={"name": my_co, "custom_is_deleted": 0}, fields=["*"], ignore_permissions=True) return frappe.get_all("SB Group Company", filters={"custom_is_deleted": 0}, fields=["*"], ignore_permissions=True) @frappe.whitelist() def save_group_company(company_name, acronym=None, code=None, status="Active", owner_user=None, notes=None, name=None): """Create or update SB Group Company (Platform Admin only).""" from sidrah_business.soft_delete import is_platform_admin if not is_platform_admin(): frappe.throw("Only Platform Administrators can manage Group Companies.", frappe.PermissionError) if name and frappe.db.exists("SB Group Company", name): doc = frappe.get_doc("SB Group Company", name) doc.company_name = company_name doc.acronym = acronym doc.code = code doc.status = status doc.owner_user = owner_user doc.notes = notes doc.save(ignore_permissions=True) else: doc = frappe.get_doc({ "doctype": "SB Group Company", "company_name": company_name, "acronym": acronym, "code": code, "status": status, "owner_user": owner_user, "notes": notes }) doc.insert(ignore_permissions=True) # When owner_user is specified, grant them SB Admin role and set their custom_group_company if owner_user and frappe.db.exists("User", owner_user): owner_doc = frappe.get_doc("User", owner_user) roles = [r.role for r in owner_doc.roles] if "SB Admin" not in roles: owner_doc.add_roles("SB Admin") frappe.db.set_value("User", owner_user, "custom_group_company", doc.name, update_modified=False) frappe.db.commit() return doc.as_dict()