2963 lines
126 KiB
Python
2963 lines
126 KiB
Python
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 '<doctype> - <workflow_state>' (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()
|